busya
ce0369ae5c
Subagents delivered tests across all uncovered backend modules: Schemas (100%): agent, auth, health, profile, settings, validation Services (98-100%): auth, profile, health, llm, mapping, resource, security, git, superset_lookup, sql_table_extractor, rbac API routes (new): auth, admin, health, environments, plugins, dashboards (helpers, projection, actions, listing), git (config, deps, env, helpers) Clean Release (100%): DTO, facade, policy_engine, stages, repos, preparation, source_isolation, compliance Git services: base, remote_providers Agent module: app, run, middleware, langgraph_setup Core: trace, cleanup, ws_log_handler, timezone, auth (config/oauth/security), matching Reports: normalizer, report_service, type_profiles Notifications: service, providers Also: - .gitignore: add .coverage, *.cover, coverage-* dirs - src/schemas/auth.py: fix AD group DN regex (comma in CN=...) - Remove co-located src/services/__tests__/ (caused pytest module collision)
261 lines
8.9 KiB
Python
261 lines
8.9 KiB
Python
# #region Test.Schemas.Settings [C:2] [TYPE Module] [SEMANTICS test,schema,settings]
|
|
# @BRIEF Tests for schemas/settings.py — NotificationChannel, ValidationPolicyBase,
|
|
# ValidationPolicyCreate, ValidationPolicyUpdate, ValidationPolicyResponse,
|
|
# TranslationScheduleItem.
|
|
# @RELATION BINDS_TO -> [SettingsSchemas]
|
|
# @TEST_EDGE: missing_field -> optional fields default correctly
|
|
# @TEST_EDGE: invalid_type -> validation rejects bad types
|
|
# @TEST_EDGE: serialization -> round-trip via model_dump/model_validate
|
|
|
|
from datetime import datetime, time, timezone
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from src.schemas.settings import (
|
|
NotificationChannel,
|
|
TranslationScheduleItem,
|
|
ValidationPolicyBase,
|
|
ValidationPolicyCreate,
|
|
ValidationPolicyResponse,
|
|
ValidationPolicyUpdate,
|
|
)
|
|
|
|
|
|
class TestNotificationChannel:
|
|
"""NotificationChannel — type and target."""
|
|
|
|
def test_minimal(self):
|
|
nc = NotificationChannel(type="SLACK", target="#alerts")
|
|
assert nc.type == "SLACK"
|
|
assert nc.target == "#alerts"
|
|
|
|
def test_missing_field_raises(self):
|
|
with pytest.raises(ValidationError):
|
|
NotificationChannel(type="SLACK")
|
|
with pytest.raises(ValidationError):
|
|
NotificationChannel(target="#alerts")
|
|
|
|
def test_serialize_roundtrip(self):
|
|
nc = NotificationChannel(type="TELEGRAM", target="chat-1")
|
|
data = nc.model_dump()
|
|
restored = NotificationChannel.model_validate(data)
|
|
assert restored.type == "TELEGRAM"
|
|
assert restored.target == "chat-1"
|
|
|
|
|
|
class TestValidationPolicyBase:
|
|
"""ValidationPolicyBase — required fields and defaults."""
|
|
|
|
def make_policy(self, **overrides):
|
|
defaults = dict(
|
|
name="Test Policy",
|
|
environment_id="env-1",
|
|
dashboard_ids=["1", "2"],
|
|
schedule_days=[0, 2, 4],
|
|
window_start=time(9, 0),
|
|
window_end=time(18, 0),
|
|
)
|
|
defaults.update(overrides)
|
|
return ValidationPolicyBase(**defaults)
|
|
|
|
def test_minimal(self):
|
|
p = self.make_policy()
|
|
assert p.name == "Test Policy"
|
|
assert p.is_active is True
|
|
assert p.notify_owners is True
|
|
assert p.alert_condition == "FAIL_ONLY"
|
|
assert p.custom_channels is None
|
|
|
|
def test_with_channels(self):
|
|
channels = [NotificationChannel(type="SLACK", target="#alerts")]
|
|
p = self.make_policy(custom_channels=channels)
|
|
assert len(p.custom_channels) == 1
|
|
assert p.custom_channels[0].type == "SLACK"
|
|
|
|
def test_inactive(self):
|
|
p = self.make_policy(is_active=False)
|
|
assert p.is_active is False
|
|
|
|
def test_alert_conditions(self):
|
|
for condition in ["FAIL_ONLY", "WARN_AND_FAIL", "ALWAYS"]:
|
|
p = self.make_policy(alert_condition=condition)
|
|
assert p.alert_condition == condition
|
|
|
|
def test_serialize_roundtrip(self):
|
|
p = self.make_policy()
|
|
data = p.model_dump()
|
|
restored = ValidationPolicyBase.model_validate(data)
|
|
assert restored.name == "Test Policy"
|
|
assert restored.is_active is True
|
|
|
|
|
|
class TestValidationPolicyCreate:
|
|
"""ValidationPolicyCreate — inherits ValidationPolicyBase."""
|
|
|
|
def test_create(self):
|
|
p = ValidationPolicyCreate(
|
|
name="New Policy",
|
|
environment_id="env-1",
|
|
dashboard_ids=["1"],
|
|
schedule_days=[1],
|
|
window_start=time(10, 0),
|
|
window_end=time(20, 0),
|
|
)
|
|
assert p.name == "New Policy"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
p = ValidationPolicyCreate(
|
|
name="P", environment_id="e1", dashboard_ids=["1"],
|
|
schedule_days=[1], window_start=time(9, 0), window_end=time(17, 0),
|
|
)
|
|
data = p.model_dump()
|
|
restored = ValidationPolicyCreate.model_validate(data)
|
|
assert restored.name == "P"
|
|
|
|
|
|
class TestValidationPolicyUpdate:
|
|
"""ValidationPolicyUpdate — all fields optional for partial update."""
|
|
|
|
def test_empty(self):
|
|
p = ValidationPolicyUpdate()
|
|
assert p.name is None
|
|
assert p.environment_id is None
|
|
assert p.is_active is None
|
|
assert p.dashboard_ids is None
|
|
assert p.schedule_days is None
|
|
assert p.window_start is None
|
|
assert p.window_end is None
|
|
assert p.notify_owners is None
|
|
assert p.custom_channels is None
|
|
assert p.alert_condition is None
|
|
|
|
def test_partial(self):
|
|
p = ValidationPolicyUpdate(name="Updated")
|
|
assert p.name == "Updated"
|
|
|
|
def test_serialize_roundtrip(self):
|
|
p = ValidationPolicyUpdate(name="Updated", is_active=False)
|
|
data = p.model_dump()
|
|
restored = ValidationPolicyUpdate.model_validate(data)
|
|
assert restored.name == "Updated"
|
|
assert restored.is_active is False
|
|
|
|
|
|
class TestValidationPolicyResponse:
|
|
"""ValidationPolicyResponse — inherits ValidationPolicyBase with id + timestamps."""
|
|
|
|
def test_minimal(self):
|
|
ts = datetime.now(timezone.utc)
|
|
p = ValidationPolicyResponse(
|
|
id="pol-1",
|
|
name="Policy",
|
|
environment_id="env-1",
|
|
dashboard_ids=["1"],
|
|
schedule_days=[1],
|
|
window_start=time(9, 0),
|
|
window_end=time(17, 0),
|
|
created_at=ts,
|
|
updated_at=ts,
|
|
)
|
|
assert p.id == "pol-1"
|
|
assert p.created_at == ts
|
|
assert p.updated_at == ts
|
|
|
|
def test_serialize_roundtrip(self):
|
|
ts = datetime.now(timezone.utc)
|
|
p = ValidationPolicyResponse(
|
|
id="pol-1", name="P", environment_id="e1", dashboard_ids=["1"],
|
|
schedule_days=[1], window_start=time(9, 0), window_end=time(17, 0),
|
|
created_at=ts, updated_at=ts,
|
|
)
|
|
data = p.model_dump()
|
|
restored = ValidationPolicyResponse.model_validate(data)
|
|
assert restored.id == "pol-1"
|
|
|
|
def test_defaults_from_base(self):
|
|
ts = datetime.now(timezone.utc)
|
|
p = ValidationPolicyResponse(
|
|
id="pol-1", name="P", environment_id="e1", dashboard_ids=["1"],
|
|
schedule_days=[1], window_start=time(9, 0), window_end=time(17, 0),
|
|
created_at=ts, updated_at=ts,
|
|
)
|
|
assert p.is_active is True
|
|
assert p.notify_owners is True
|
|
assert p.alert_condition == "FAIL_ONLY"
|
|
|
|
def test_from_attributes(self):
|
|
ts = datetime.now(timezone.utc)
|
|
class FakeORM:
|
|
id = "pol-1"
|
|
name = "P"
|
|
environment_id = "e1"
|
|
is_active = True
|
|
dashboard_ids = ["1"]
|
|
schedule_days = [1]
|
|
window_start = time(9, 0)
|
|
window_end = time(17, 0)
|
|
notify_owners = False
|
|
custom_channels = None
|
|
alert_condition = "WARN_AND_FAIL"
|
|
created_at = ts
|
|
updated_at = ts
|
|
p = ValidationPolicyResponse.model_validate(FakeORM())
|
|
assert p.notify_owners is False
|
|
|
|
|
|
class TestTranslationScheduleItem:
|
|
"""TranslationScheduleItem — schedule with cron expression."""
|
|
|
|
def test_minimal(self):
|
|
s = TranslationScheduleItem(
|
|
schedule_id="s-1", job_id="j-1",
|
|
cron_expression="0 */6 * * *", timezone="UTC", is_active=True,
|
|
)
|
|
assert s.job_name is None
|
|
assert s.execution_mode == "full"
|
|
assert s.last_run_at is None
|
|
assert s.next_run_at is None
|
|
assert s.created_by is None
|
|
assert s.created_at is None
|
|
|
|
def test_with_all_fields(self):
|
|
ts = datetime.now(timezone.utc)
|
|
s = TranslationScheduleItem(
|
|
schedule_id="s-1", job_id="j-1", job_name="Nightly Sync",
|
|
cron_expression="0 0 * * *", timezone="America/New_York",
|
|
is_active=False, execution_mode="incremental",
|
|
last_run_at=ts, next_run_at=ts, created_by="admin", created_at=ts,
|
|
)
|
|
assert s.job_name == "Nightly Sync"
|
|
assert s.execution_mode == "incremental"
|
|
assert s.is_active is False
|
|
|
|
def test_serialize_roundtrip(self):
|
|
s = TranslationScheduleItem(
|
|
schedule_id="s-1", job_id="j-1",
|
|
cron_expression="0 0 * * *", timezone="UTC", is_active=True,
|
|
)
|
|
data = s.model_dump()
|
|
restored = TranslationScheduleItem.model_validate(data)
|
|
assert restored.schedule_id == "s-1"
|
|
assert restored.is_active is True
|
|
|
|
def test_from_attributes(self):
|
|
ts = datetime.now(timezone.utc)
|
|
class FakeORM:
|
|
schedule_id = "s-1"
|
|
job_id = "j-1"
|
|
job_name = "Test"
|
|
cron_expression = "*/5 * * * *"
|
|
timezone = "UTC"
|
|
is_active = True
|
|
execution_mode = "full"
|
|
last_run_at = ts
|
|
next_run_at = ts
|
|
created_by = "admin"
|
|
created_at = ts
|
|
s = TranslationScheduleItem.model_validate(FakeORM())
|
|
assert s.cron_expression == "*/5 * * * *"
|
|
assert s.execution_mode == "full"
|
|
# #endregion Test.Schemas.Settings
|