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.
100 lines
4.4 KiB
100 lines
4.4 KiB
from datetime import time
|
|
from django.urls import reverse
|
|
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):
|
|
response = self.client.delete(self.detail_url)
|
|
self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
|
|
self.assertFalse(CrawlTask.objects.filter(id=self.task.id).exists())
|
|
|
|
def test_trigger_crawl_task(self):
|
|
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)
|
|
|
|
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')
|