You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
205 lines
8.2 KiB
205 lines
8.2 KiB
from datetime import time
|
|
from django.urls import reverse
|
|
from unittest.mock import patch
|
|
from rest_framework import status
|
|
from rest_framework.test import APITestCase
|
|
from .models import CrawlTask, CrawlRun
|
|
|
|
class HealthCheckTests(APITestCase):
|
|
def test_health_check_endpoint(self):
|
|
url = reverse('health-check')
|
|
response = self.client.get(url)
|
|
self.assertIn(response.status_code, [status.HTTP_200_OK, status.HTTP_503_SERVICE_UNAVAILABLE])
|
|
self.assertIn('status', response.data)
|
|
self.assertIn('database', response.data)
|
|
self.assertIn('redis', response.data)
|
|
|
|
|
|
class CrawlTaskAPITests(APITestCase):
|
|
def setUp(self):
|
|
self.task_data = {
|
|
"title": "Tehran Apartments",
|
|
"divar_url": "https://divar.ir/s/tehran/buy-apartment",
|
|
"detection_prompt": "Find urgent sales",
|
|
"interval_minutes": 15,
|
|
"start_hour": "08:00:00",
|
|
"end_hour": "23:00:00",
|
|
"telegram_channel_id": "@my_channel",
|
|
"is_active": True
|
|
}
|
|
self.task = CrawlTask.objects.create(
|
|
title="Existing Task",
|
|
divar_url="https://divar.ir/s/tehran/rent-apartment",
|
|
detection_prompt="Find immediate rent options",
|
|
interval_minutes=30,
|
|
start_hour=time(9, 0),
|
|
end_hour=time(22, 0),
|
|
telegram_channel_id="@rent_channel",
|
|
is_active=True
|
|
)
|
|
self.list_url = "/api/crawlers/"
|
|
self.detail_url = f"/api/crawlers/{self.task.id}/"
|
|
self.trigger_url = f"/api/crawlers/{self.task.id}/trigger/"
|
|
self.runs_url = f"/api/crawlers/{self.task.id}/runs/"
|
|
|
|
def test_create_crawl_task_success(self):
|
|
response = self.client.post(self.list_url, self.task_data, format='json')
|
|
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
|
|
self.assertEqual(response.data['title'], self.task_data['title'])
|
|
self.assertEqual(response.data['divar_url'], self.task_data['divar_url'])
|
|
|
|
def test_create_crawl_task_invalid_url(self):
|
|
invalid_data = self.task_data.copy()
|
|
invalid_data['divar_url'] = "https://invalid-url.com/s/tehran"
|
|
response = self.client.post(self.list_url, invalid_data, format='json')
|
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
|
self.assertIn('divar_url', response.data)
|
|
|
|
def test_get_crawl_tasks_list(self):
|
|
response = self.client.get(self.list_url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(len(response.data), 1)
|
|
self.assertEqual(response.data[0]['title'], self.task.title)
|
|
|
|
def test_get_crawl_task_detail(self):
|
|
response = self.client.get(self.detail_url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['title'], self.task.title)
|
|
|
|
def test_update_crawl_task(self):
|
|
update_data = {"title": "Updated Task Name"}
|
|
response = self.client.patch(self.detail_url, update_data, format='json')
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(response.data['title'], "Updated Task Name")
|
|
|
|
def test_delete_crawl_task(self):
|
|
# Create a run and evaluation for the task to verify cascade deletion
|
|
run = CrawlRun.objects.create(
|
|
crawl_task=self.task,
|
|
status='SUCCESS'
|
|
)
|
|
from ads.models import Ad, AdEvaluation
|
|
ad = Ad.objects.create(
|
|
divar_token="delete_token",
|
|
title="Delete Test Ad",
|
|
description="Desc",
|
|
url="https://divar.ir/v/delete_token"
|
|
)
|
|
evaluation = AdEvaluation.objects.create(
|
|
crawl_task=self.task,
|
|
ad=ad,
|
|
is_flagged=True
|
|
)
|
|
|
|
# Ensure they exist initially
|
|
self.assertTrue(CrawlRun.objects.filter(id=run.id).exists())
|
|
self.assertTrue(AdEvaluation.objects.filter(id=evaluation.id).exists())
|
|
|
|
response = self.client.delete(self.detail_url)
|
|
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
|
|
|
|
# Verify cascade deletion
|
|
self.assertFalse(CrawlTask.objects.filter(id=self.task.id).exists())
|
|
self.assertFalse(CrawlRun.objects.filter(id=run.id).exists())
|
|
self.assertFalse(AdEvaluation.objects.filter(id=evaluation.id).exists())
|
|
|
|
@patch('crawler.tasks.run_crawl_pipeline.delay')
|
|
def test_trigger_crawl_task(self, mock_delay):
|
|
runs_before = CrawlRun.objects.filter(crawl_task=self.task).count()
|
|
response = self.client.post(self.trigger_url)
|
|
self.assertEqual(response.status_code, status.HTTP_202_ACCEPTED)
|
|
self.assertIn('run_id', response.data)
|
|
self.assertIn('status', response.data)
|
|
runs_after = CrawlRun.objects.filter(crawl_task=self.task).count()
|
|
self.assertEqual(runs_after, runs_before + 1)
|
|
mock_delay.assert_called_once()
|
|
|
|
def test_get_crawl_task_runs(self):
|
|
run = CrawlRun.objects.create(
|
|
crawl_task=self.task,
|
|
status='SUCCESS',
|
|
ads_fetched_count=5,
|
|
ads_evaluated_count=5,
|
|
ads_flagged_count=1
|
|
)
|
|
response = self.client.get(self.runs_url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
self.assertEqual(len(response.data), 1)
|
|
self.assertEqual(response.data[0]['id'], str(run.id))
|
|
self.assertEqual(response.data[0]['status'], 'SUCCESS')
|
|
|
|
def test_delete_crawl_run(self):
|
|
run = CrawlRun.objects.create(
|
|
crawl_task=self.task,
|
|
status='SUCCESS'
|
|
)
|
|
delete_url = f"/api/crawlers/runs/{run.id}/"
|
|
response = self.client.delete(delete_url)
|
|
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
|
|
self.assertFalse(CrawlRun.objects.filter(id=run.id).exists())
|
|
|
|
def test_stop_crawl_run(self):
|
|
run = CrawlRun.objects.create(
|
|
crawl_task=self.task,
|
|
status='RUNNING'
|
|
)
|
|
stop_url = f"/api/crawlers/runs/{run.id}/stop/"
|
|
response = self.client.post(stop_url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
run.refresh_from_db()
|
|
self.assertEqual(run.status, 'FAILED')
|
|
self.assertIn("متوقف شد", run.error_log)
|
|
|
|
|
|
|
|
class CrawlTaskCeleryTests(APITestCase):
|
|
def setUp(self):
|
|
self.task = CrawlTask.objects.create(
|
|
title="Tehran Apartments",
|
|
divar_url="https://divar.ir/s/tehran/buy-apartment?query=urgent",
|
|
detection_prompt="Find urgent sales",
|
|
interval_minutes=15,
|
|
start_hour=time(8, 0),
|
|
end_hour=time(23, 0)
|
|
)
|
|
self.run = CrawlRun.objects.create(
|
|
crawl_task=self.task,
|
|
status='RUNNING'
|
|
)
|
|
|
|
def test_extract_price_from_widget(self):
|
|
from crawler.tasks import extract_price_from_widget
|
|
|
|
# Test direct price_text
|
|
widget1 = {"data": {"title": "Apartment", "price_text": "۴,۵۰۰,۰۰۰ تومان"}}
|
|
self.assertEqual(extract_price_from_widget(widget1), "۴,۵۰۰,۰۰۰ تومان")
|
|
|
|
# Test middle_description_text
|
|
widget2 = {"data": {"title": "Car", "middle_description_text": "۱۲,۰۰۰,۰۰۰ تومان"}}
|
|
self.assertEqual(extract_price_from_widget(widget2), "۱۲,۰۰۰,۰۰۰ تومان")
|
|
|
|
# Test price dict
|
|
widget3 = {"data": {"title": "Item", "price": {"text": "توافقی"}}}
|
|
self.assertEqual(extract_price_from_widget(widget3), "توافقی")
|
|
|
|
def test_run_crawl_pipeline_inactive_task(self):
|
|
self.task.is_active = False
|
|
self.task.save()
|
|
|
|
from crawler.tasks import run_crawl_pipeline
|
|
run_crawl_pipeline(self.run.id)
|
|
|
|
self.run.refresh_from_db()
|
|
self.assertEqual(self.run.status, 'FAILED')
|
|
self.assertEqual(self.run.error_log, "CrawlTask is inactive.")
|
|
|
|
@patch('crawler.tasks.run_crawl_pipeline.delay')
|
|
def test_check_and_schedule_crawl_tasks(self, mock_delay):
|
|
from crawler.tasks import check_and_schedule_crawl_tasks
|
|
CrawlRun.objects.all().delete()
|
|
|
|
check_and_schedule_crawl_tasks()
|
|
|
|
self.assertEqual(CrawlRun.objects.count(), 1)
|
|
mock_delay.assert_called_once()
|
|
|