48 lines
2.1 KiB
Python
48 lines
2.1 KiB
Python
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||
|
|
|
||
|
|
from app.modules.internet_connections.backend import change_case_service as service
|
||
|
|
|
||
|
|
|
||
|
|
def test_relevant_change_filter_excludes_manual_classification_and_customer():
|
||
|
|
changes = service.filter_relevant_changes({
|
||
|
|
'monthly_cost': {'from': 100, 'to': 125},
|
||
|
|
'customer_id': {'from': None, 'to': 7},
|
||
|
|
'allocation_model': {'from': 'dedicated', 'to': 'shared'},
|
||
|
|
'notes': {'from': 'før', 'to': 'efter'},
|
||
|
|
})
|
||
|
|
assert changes == {'monthly_cost': {'from': 100, 'to': 125}}
|
||
|
|
|
||
|
|
|
||
|
|
def test_same_import_merges_into_existing_case(monkeypatch):
|
||
|
|
writes = []
|
||
|
|
monkeypatch.setattr(service, 'execute_query_single', lambda query, params=None: (
|
||
|
|
{'id': 4, 'sag_id': 91, 'changes': {'monthly_cost': {'from': 100, 'to': 125}}}
|
||
|
|
if 'FROM internet_connection_change_cases' in query else None
|
||
|
|
))
|
||
|
|
monkeypatch.setattr(service, 'execute_query', lambda query, params=None, fetch=True: writes.append((query, params)))
|
||
|
|
|
||
|
|
result = service.ensure_external_change_case(
|
||
|
|
connection_id=5, source_type='test_import', source_key='same-file',
|
||
|
|
source_label='Testfil', connection_name='Fiber', reference='NKA-1', provider='Leverandør',
|
||
|
|
owner_customer_id=7, changes={'speed_mbps': {'from': 100, 'to': 200}},
|
||
|
|
)
|
||
|
|
|
||
|
|
assert result['case_id'] == 91
|
||
|
|
assert result['created'] is False
|
||
|
|
assert set(result['changes']) == {'monthly_cost', 'speed_mbps'}
|
||
|
|
assert any('UPDATE sag_sager' in query for query, _ in writes)
|
||
|
|
|
||
|
|
|
||
|
|
def test_case_failure_is_reported_without_raising(monkeypatch):
|
||
|
|
monkeypatch.setattr(service, 'execute_query_single', lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError('database down')))
|
||
|
|
monkeypatch.setattr(service, 'execute_query', lambda *args, **kwargs: None)
|
||
|
|
result = service.ensure_external_change_case(
|
||
|
|
connection_id=5, source_type='test', source_key='run-1', source_label='Test',
|
||
|
|
changes={'status': {'from': 'active', 'to': 'inactive'}},
|
||
|
|
)
|
||
|
|
assert result['case_id'] is None
|
||
|
|
assert result['error'] == 'database down'
|