PouyaKhajavi
/
divar-crawler
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

test(backend): add unit tests for health check, crawler tasks, and ads evaluations

master
PouyaKhajavi 2 days ago
parent
12c6cefb22
commit
b97950391b
2 changed files with 195 additions and 4 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 98
      backend/ads/tests.py
  2. 101
      backend/crawler/tests.py

98
backend/ads/tests.py
View File

@ -1,3 +1,97 @@
from django.test import TestCase
from datetime import time
from rest_framework import status
from rest_framework.test import APITestCase
from crawler.models import CrawlTask
from .models import Ad, AdEvaluation
# Create your tests here.
class AdEvaluationAPITests(APITestCase):
def setUp(self):
self.task1 = CrawlTask.objects.create(
title="Task 1",
divar_url="https://divar.ir/s/tehran/buy-apartment",
detection_prompt="Prompt 1",
interval_minutes=15,
start_hour=time(8, 0),
end_hour=time(23, 0)
)
self.task2 = CrawlTask.objects.create(
title="Task 2",
divar_url="https://divar.ir/s/tehran/rent-apartment",
detection_prompt="Prompt 2",
interval_minutes=30,
start_hour=time(9, 0),
end_hour=time(22, 0)
)
self.ad1 = Ad.objects.create(
divar_token="token1",
title="Apartment in Tehran",
description="Beautiful apartment",
price="10,000,000,000",
category="buy-apartment",
images=["http://image1.jpg"],
url="https://divar.ir/v/token1"
)
self.ad2 = Ad.objects.create(
divar_token="token2",
title="Rent in Tehran",
description="Nice place",
price="500,000,000",
category="rent-apartment",
images=[],
url="https://divar.ir/v/token2"
)
self.eval1 = AdEvaluation.objects.create(
crawl_task=self.task1,
ad=self.ad1,
is_flagged=True,
reason="Matches perfectly",
confidence=0.95,
extracted_fields={"urgency": "high"}
)
self.eval2 = AdEvaluation.objects.create(
crawl_task=self.task2,
ad=self.ad2,
is_flagged=False,
reason="Does not match",
confidence=0.1,
extracted_fields={}
)
self.list_url = "/api/ads/"
def test_get_evaluations_list(self):
response = self.client.get(self.list_url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 2)
# Check that ad details are nested
self.assertEqual(response.data[0]['ad']['divar_token'], self.ad2.divar_token) # Ordered by -evaluated_at
self.assertEqual(response.data[1]['ad']['divar_token'], self.ad1.divar_token)
def test_filter_by_crawl_task(self):
response = self.client.get(self.list_url, {"crawl_task": self.task1.id})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['id'], str(self.eval1.id))
def test_filter_by_is_flagged_true(self):
response = self.client.get(self.list_url, {"is_flagged": "true"})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['id'], str(self.eval1.id))
self.assertTrue(response.data[0]['is_flagged'])
def test_filter_by_is_flagged_false(self):
response = self.client.get(self.list_url, {"is_flagged": "false"})
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(len(response.data), 1)
self.assertEqual(response.data[0]['id'], str(self.eval2.id))
self.assertFalse(response.data[0]['is_flagged'])
def test_get_evaluation_detail(self):
detail_url = f"{self.list_url}{self.eval1.id}/"
response = self.client.get(detail_url)
self.assertEqual(response.status_code, status.HTTP_200_OK)
self.assertEqual(response.data['id'], str(self.eval1.id))
self.assertEqual(response.data['ad']['title'], self.ad1.title)

101
backend/crawler/tests.py
View File

@ -1,3 +1,100 @@
from django.test import TestCase
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
# Create your tests here.
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')
Write Preview
Loading…
Cancel
Save