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_parse_divar_url(self): from crawler.tasks import parse_divar_url city, category, query_params = parse_divar_url(self.task.divar_url) self.assertEqual(city, 'tehran') self.assertEqual(category, 'buy-apartment') self.assertEqual(query_params, {'query': 'urgent'}) 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()