Aleksandr Meshchriakov
45bca018b5
Some checks failed
CI/CD Pipeline / Code Quality Checks (pull_request) Successful in 3m6s
CI/CD Pipeline / Run API Inventory E2E Tests (pull_request) Has been cancelled
CI/CD Pipeline / Telegram Notify Success (pull_request) Has been cancelled
CI/CD Pipeline / Run Tests (pull_request) Has been cancelled
104 lines
3.6 KiB
Python
104 lines
3.6 KiB
Python
"""Tests for exchange serializers."""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
from apps.exchange.serializers import (
|
|
ExchangeCopyRequestSerializer,
|
|
ExchangePeriodicTaskUpsertSerializer,
|
|
)
|
|
from django.test import SimpleTestCase
|
|
|
|
|
|
class ExchangeCopyRequestSerializerTest(SimpleTestCase):
|
|
def test_single_mode_requires_table(self):
|
|
serializer = ExchangeCopyRequestSerializer(data={"mode": "single"})
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertIn("table", serializer.errors)
|
|
|
|
def test_selected_mode_requires_tables(self):
|
|
serializer = ExchangeCopyRequestSerializer(data={"mode": "selected"})
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertIn("tables", serializer.errors)
|
|
|
|
def test_table_and_tables_are_rejected_for_wrong_modes(self):
|
|
serializer_with_table = ExchangeCopyRequestSerializer(
|
|
data={"mode": "all", "table": "parsers_manufacturer"}
|
|
)
|
|
self.assertFalse(serializer_with_table.is_valid())
|
|
self.assertIn("table", serializer_with_table.errors)
|
|
|
|
serializer_with_tables = ExchangeCopyRequestSerializer(
|
|
data={"mode": "all", "tables": ["parsers_manufacturer"]}
|
|
)
|
|
self.assertFalse(serializer_with_tables.is_valid())
|
|
self.assertIn("tables", serializer_with_tables.errors)
|
|
|
|
|
|
class ExchangePeriodicTaskUpsertSerializerTest(SimpleTestCase):
|
|
def test_interval_schedule_requires_fields(self):
|
|
serializer = ExchangePeriodicTaskUpsertSerializer(
|
|
data={"schedule_type": "interval"}
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertIn("interval_every", serializer.errors)
|
|
self.assertIn("interval_period", serializer.errors)
|
|
|
|
def test_daily_schedule_requires_fields(self):
|
|
serializer = ExchangePeriodicTaskUpsertSerializer(
|
|
data={"schedule_type": "daily"}
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertIn("crontab_minute", serializer.errors)
|
|
self.assertIn("crontab_hour", serializer.errors)
|
|
|
|
def test_partial_update_preserves_existing_payload(self):
|
|
instance = SimpleNamespace(
|
|
name="copy-job",
|
|
interval_id=1,
|
|
interval=SimpleNamespace(every=5, period="minutes"),
|
|
crontab_id=None,
|
|
kwargs='{"payload": {"mode": "single", "table": "old_table"}}',
|
|
)
|
|
serializer = ExchangePeriodicTaskUpsertSerializer(
|
|
instance,
|
|
data={},
|
|
partial=True,
|
|
)
|
|
|
|
self.assertTrue(serializer.is_valid(), serializer.errors)
|
|
self.assertEqual(
|
|
serializer.validated_data["payload"],
|
|
{
|
|
"mode": "single",
|
|
"table": "old_table",
|
|
"tables": None,
|
|
"truncate_before_copy": True,
|
|
"notify_on_error": False,
|
|
},
|
|
)
|
|
|
|
def test_invalid_schedule_type_is_rejected(self):
|
|
serializer = ExchangePeriodicTaskUpsertSerializer(
|
|
data={
|
|
"schedule_type": "crontab",
|
|
}
|
|
)
|
|
|
|
self.assertFalse(serializer.is_valid())
|
|
self.assertIn("schedule_type", serializer.errors)
|
|
|
|
def test_notify_on_error_is_added_to_payload(self):
|
|
serializer = ExchangePeriodicTaskUpsertSerializer(
|
|
data={
|
|
"schedule_type": "interval",
|
|
"interval_every": 1,
|
|
"interval_period": "hours",
|
|
"notify_on_error": True,
|
|
}
|
|
)
|
|
|
|
self.assertTrue(serializer.is_valid(), serializer.errors)
|
|
self.assertTrue(serializer.validated_data["payload"]["notify_on_error"])
|