1cfe5aee76
- Created tables for AI benchmark runs and results to facilitate model evaluation. - Added expected answers column to benchmark results. - Introduced tables for internet connection change cases and vTiger archive management, including records, relations, and checkpoints. - Implemented triggers to enforce append-only behavior for vTiger archive records and files. - Enhanced solution management with soft delete capabilities for sag_solutions and knowledge_articles. feat(scripts): add CRM benchmarking script for Ollama models - Developed a Python script to benchmark CRM models exposed through Ollama, including various test cases and scoring mechanisms. test(tests): add comprehensive tests for new features - Implemented tests for internet change case service, vTiger archive functionality, and sag solution knowledge management. - Ensured coverage for edge cases and error handling in the new features.
258 lines
11 KiB
Python
258 lines
11 KiB
Python
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from app.admin import router
|
|
from app.admin import vtiger_archive as archive
|
|
from app.admin import hub_impact
|
|
from app.admin import archive_bundle
|
|
from app.services.vtiger_service import VTigerService
|
|
|
|
|
|
def test_hidden_admin_dependency_returns_404_for_non_superadmin():
|
|
with pytest.raises(HTTPException) as exc:
|
|
router.require_hidden_superadmin({'username': 'employee', 'is_superadmin': False})
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
def test_archive_record_is_append_only_and_deduplicated(monkeypatch):
|
|
writes = []
|
|
previous = {'revision_no': 1, 'payload_sha256': 'different'}
|
|
monkeypatch.setattr(archive, 'execute_query_single', lambda query, params=None: previous if 'payload_sha256' in query else None)
|
|
monkeypatch.setattr(archive, 'execute_query', lambda query, params=None, fetch=True: writes.append((query, params)))
|
|
inserted = archive.archive_record(3, 'Accounts', {
|
|
'id': '3x44', 'accountname': 'Test', 'contact_id': '4x99', 'modifiedtime': '2026-08-30 10:00:00',
|
|
})
|
|
assert inserted is True
|
|
assert any('INSERT INTO vtiger_archive_records' in sql for sql, _ in writes)
|
|
assert any('INSERT INTO vtiger_archive_relations' in sql for sql, _ in writes)
|
|
|
|
|
|
def test_unchanged_payload_does_not_create_revision(monkeypatch):
|
|
record = {'id': '3x44', 'accountname': 'Test'}
|
|
_, digest = archive._canonical_payload(record)
|
|
monkeypatch.setattr(archive, 'execute_query_single', lambda *args, **kwargs: {'revision_no': 2, 'payload_sha256': digest})
|
|
monkeypatch.setattr(archive, 'execute_query', lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError('must not write')))
|
|
assert archive.archive_record(4, 'Accounts', record) is False
|
|
|
|
|
|
def test_readiness_requires_successful_approved_final(monkeypatch):
|
|
monkeypatch.setattr(archive, 'execute_query_single', lambda *args, **kwargs: {
|
|
'id': 9, 'status': 'completed', 'critical_errors': [], 'control_approved_at': '2026-08-30',
|
|
})
|
|
assert archive.termination_readiness()['ready'] is True
|
|
|
|
monkeypatch.setattr(archive, 'execute_query_single', lambda *args, **kwargs: None)
|
|
result = archive.termination_readiness()
|
|
assert result['ready'] is False
|
|
assert len(result['reasons']) == 2
|
|
|
|
|
|
def test_failed_vtiger_query_is_not_treated_as_empty_module(monkeypatch):
|
|
class Service:
|
|
last_query_error = {'message': 'rate limited'}
|
|
last_query_status = 429
|
|
async def query(self, query):
|
|
return []
|
|
monkeypatch.setattr(archive, 'get_vtiger_service', lambda: Service())
|
|
async def no_sleep(_seconds):
|
|
return None
|
|
monkeypatch.setattr(archive.asyncio, 'sleep', no_sleep)
|
|
with pytest.raises(RuntimeError):
|
|
asyncio.run(archive._fetch_module('Accounts'))
|
|
|
|
|
|
def test_failed_vtiger_count_cannot_pass_full_archive_control(monkeypatch):
|
|
class Service:
|
|
last_query_error = {'message': 'timeout'}
|
|
last_query_status = None
|
|
|
|
async def query(self, query):
|
|
return []
|
|
|
|
async def no_sleep(_seconds):
|
|
return None
|
|
|
|
monkeypatch.setattr(archive, 'get_vtiger_service', lambda: Service())
|
|
monkeypatch.setattr(archive.asyncio, 'sleep', no_sleep)
|
|
with pytest.raises(RuntimeError, match='kontrollere antal poster'):
|
|
asyncio.run(archive._source_module_count('Accounts'))
|
|
|
|
|
|
def test_vtiger_file_retrieve_accepts_list_result(monkeypatch):
|
|
import base64
|
|
|
|
class Response:
|
|
status = 200
|
|
async def __aenter__(self): return self
|
|
async def __aexit__(self, *args): return None
|
|
async def json(self, content_type=None):
|
|
return {'success': True, 'result': [{
|
|
'fileid': '7x2', 'filename': 'test.txt', 'filecontents': base64.b64encode(b'hello').decode(),
|
|
}]}
|
|
|
|
class Session:
|
|
async def __aenter__(self): return self
|
|
async def __aexit__(self, *args): return None
|
|
def get(self, *args, **kwargs): return Response()
|
|
|
|
monkeypatch.setattr('app.services.vtiger_service.aiohttp.ClientSession', Session)
|
|
service = VTigerService()
|
|
service.rest_endpoint = 'https://example.invalid'
|
|
service.api_key = 'key'
|
|
service.username = 'user'
|
|
result = asyncio.run(service.retrieve_file('7x2'))
|
|
assert result['content'] == b'hello'
|
|
|
|
|
|
def test_vtiger_archive_uses_offset_pagination(monkeypatch):
|
|
queries = []
|
|
class Service:
|
|
last_query_error = None
|
|
last_query_status = 200
|
|
async def query(self, query):
|
|
queries.append(query)
|
|
if 'LIMIT 0, 100' in query:
|
|
return [{'id': f'3x{i}'} for i in range(100)]
|
|
if 'LIMIT 100, 100' in query:
|
|
return [{'id': '3x101'}, {'id': '3x102'}]
|
|
return []
|
|
async def no_sleep(_seconds):
|
|
return None
|
|
monkeypatch.setattr(archive, 'get_vtiger_service', lambda: Service())
|
|
monkeypatch.setattr(archive.asyncio, 'sleep', no_sleep)
|
|
rows = asyncio.run(archive._fetch_module('Accounts'))
|
|
assert len(rows) == 102
|
|
assert 'LIMIT 0, 100' in queries[0]
|
|
assert 'LIMIT 100, 100' in queries[1]
|
|
assert 'LIMIT 102, 100' in queries[2]
|
|
|
|
|
|
def test_project_ct_migration_enforces_immutable_revisions():
|
|
sql = Path('migrations/1030_project_ct_vtiger_archive.sql').read_text()
|
|
assert 'vtiger_archive_versions' in sql
|
|
assert 'vtiger_archive_records' in sql
|
|
assert 'vtiger_archive_relations' in sql
|
|
assert 'vtiger_archive_checkpoints' in sql
|
|
assert 'hub_impact_reports' in sql
|
|
assert 'BEFORE UPDATE OR DELETE ON vtiger_archive_records' in sql
|
|
file_sql = Path('migrations/1032_project_ct_archive_files.sql').read_text()
|
|
assert 'vtiger_archive_files' in file_sql
|
|
assert 'BEFORE UPDATE OR DELETE ON vtiger_archive_files' in file_sql
|
|
assert 'source_module' in Path('migrations/1033_project_ct_archive_file_sources.sql').read_text()
|
|
|
|
|
|
def test_impact_periods_must_be_equal(monkeypatch):
|
|
from datetime import date
|
|
with pytest.raises(ValueError, match='lige lange'):
|
|
hub_impact.build_impact_report(1, date(2025, 1, 1), date(2025, 1, 10),
|
|
date(2026, 1, 1), date(2026, 1, 9))
|
|
|
|
|
|
def test_vtiger_employee_matching_and_long_entry_exclusion():
|
|
from datetime import date
|
|
records = [
|
|
{'module': 'Users', 'vtiger_id': '19x1', 'is_deleted': False,
|
|
'payload': {'id': '19x1', 'email1': 'TECH@EXAMPLE.COM', 'first_name': 'Old', 'last_name': 'Name'}},
|
|
{'module': 'Timelog', 'vtiger_id': '36x1', 'is_deleted': False,
|
|
'payload': {'createdtime': '2025-01-03 10:00:00', 'assigned_user_id': '19x1', 'time_spent': '02:30'}},
|
|
{'module': 'Timelog', 'vtiger_id': '36x2', 'is_deleted': False,
|
|
'payload': {'createdtime': '2025-01-04 10:00:00', 'assigned_user_id': '19x1', 'hours': '17'}},
|
|
{'module': 'Cases', 'vtiger_id': '17x3', 'is_deleted': False,
|
|
'payload': {'createdtime': '2025-01-05 10:00:00', 'assigned_user_id': '19x1'}},
|
|
]
|
|
result = hub_impact._finalize(
|
|
hub_impact._vtiger_metrics(records, date(2025, 1, 1), date(2025, 1, 10), {
|
|
'tech@example.com': {'user_id': 7, 'email': 'tech@example.com', 'full_name': 'Hub Technician'}
|
|
}), date(2025, 1, 1), date(2025, 1, 10),
|
|
)
|
|
assert result['totals']['hours'] == 2.5
|
|
assert result['totals']['time_entries'] == 1
|
|
assert result['totals']['cases'] == 1
|
|
assert result['employees'][0]['key'] == 'hub:7'
|
|
assert result['anomalies'][0]['type'] == 'over_16_hours'
|
|
|
|
|
|
def test_hub_active_timer_is_anomaly_not_kpi(monkeypatch):
|
|
from datetime import date
|
|
def fake_query(sql, params=None):
|
|
if 'FROM tmodule_times' in sql:
|
|
return [{'id': 1, 'medarbejder_id': 4, 'aktiv_timer': True, 'hours': 8},
|
|
{'id': 2, 'medarbejder_id': 4, 'aktiv_timer': False, 'hours': 2}]
|
|
return []
|
|
monkeypatch.setattr(hub_impact, 'execute_query', fake_query)
|
|
metrics = hub_impact._finalize(hub_impact._hub_metrics(
|
|
date(2026, 1, 1), date(2026, 1, 2),
|
|
{4: {'user_id': 4, 'email': 'a@example.com', 'full_name': 'A'}},
|
|
), date(2026, 1, 1), date(2026, 1, 2))
|
|
assert metrics['totals']['hours'] == 2
|
|
assert metrics['totals']['time_entries'] == 1
|
|
assert metrics['anomalies'][0]['type'] == 'active_timer'
|
|
|
|
|
|
def test_excel_export_contains_required_sheets():
|
|
from app.admin.router import _impact_workbook
|
|
load_workbook = pytest.importorskip('openpyxl').load_workbook
|
|
import io
|
|
result = {
|
|
'archive_version': {'id': 2},
|
|
'periods': {'vtiger': {'from': '2025-01-01', 'to': '2025-01-31'}, 'hub': {'from': '2026-01-01', 'to': '2026-01-31'}},
|
|
'comparison': {'hours': 1, 'time_entries': 2, 'cases': 3, 'orders': 4},
|
|
'vtiger': {'totals': {'hours': 1, 'time_entries': 1, 'cases': 1, 'orders': 1}, 'employees': [], 'anomalies': [], 'data_quality': {'missing_dates': 0, 'unmatched_employees': []}},
|
|
'hub': {'totals': {'hours': 2, 'time_entries': 3, 'cases': 4, 'orders': 5}, 'employees': [], 'anomalies': [], 'data_quality': {'missing_dates': 0, 'unmatched_employees': []}},
|
|
}
|
|
workbook = load_workbook(io.BytesIO(_impact_workbook(result)))
|
|
assert workbook.sheetnames == ['Overblik', 'Effekt', 'Pr medarbejder', 'Afvigelser', 'Datakvalitet']
|
|
|
|
|
|
def test_secret_page_and_profile_link_are_registered():
|
|
assert '@router.get("/admin/hub-impact"' in Path('app/admin/views.py').read_text()
|
|
base = Path('app/shared/frontend/base.html').read_text()
|
|
assert 'projectCtProfileLink' in base
|
|
assert "user.is_superadmin === true" in base
|
|
|
|
|
|
def test_bundle_checksum_rejects_changed_content(tmp_path):
|
|
import hashlib
|
|
import json
|
|
import zipfile
|
|
bundle = tmp_path / 'archive.zip'
|
|
with zipfile.ZipFile(bundle, 'w') as archive_file:
|
|
archive_file.writestr('records.jsonl', b'{"id":1}\n')
|
|
archive_file.writestr('manifest.json', json.dumps({
|
|
'schema_version': 1,
|
|
'entries': {'records.jsonl': {'sha256': hashlib.sha256(b'other').hexdigest()}},
|
|
}))
|
|
with zipfile.ZipFile(bundle) as archive_file:
|
|
manifest = json.loads(archive_file.read('manifest.json'))
|
|
with pytest.raises(ValueError, match='Checksumfejl'):
|
|
archive_bundle._verify_entry(archive_file, manifest, 'records.jsonl')
|
|
|
|
|
|
def test_bundle_transfer_contains_original_files_and_never_uses_vtiger():
|
|
source = Path('app/admin/archive_bundle.py').read_text()
|
|
assert 'document_policy' in source
|
|
assert 'metadata_and_original_bytes' in source
|
|
assert 'files.jsonl' in source
|
|
assert 'contacted_vtiger": False' in source
|
|
assert 'vtiger_service' not in source
|
|
router_source = Path('app/admin/router.py').read_text()
|
|
assert '/admin/vtiger-archive/bundle.zip' in router_source
|
|
assert '/admin/vtiger-archive/import-bundle' in router_source
|
|
|
|
|
|
def test_productivity_exposes_clear_daily_efficiency_metrics():
|
|
from datetime import date
|
|
metrics = {'employees': {
|
|
'hub:1': {'key': 'hub:1', 'name': 'A', 'email': 'a@example.com', 'hours': hub_impact.Decimal('8'),
|
|
'time_entries': 4, 'cases': 2, 'orders': 1}},
|
|
'anomalies': [], 'data_quality': {'unmatched_employees': [], 'missing_dates': 0}}
|
|
result = hub_impact._finalize(metrics, date(2026, 8, 31), date(2026, 8, 31))
|
|
assert result['totals']['productivity']['deliveries_per_workday'] == 7
|
|
assert result['totals']['productivity']['hours_per_case_or_order'] == 2.67
|