divar-crawler/backend/crawler/tests.py

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())


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.")