bmc_hub/tests/test_sag_module.py

from datetime import datetime
from io import BytesIO
from pathlib import Path
import asyncio
import re
import sys
import types

import pytest
from fastapi import HTTPException, UploadFile, Request

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from app.utils.safe_html import sanitize_safe_html

# Keep these focused unit tests independent of optional auth runtime packages.
auth_dependencies_stub = types.ModuleType("app.core.auth_dependencies")


def _allow_test_user(*_permissions):
    async def dependency(_request: Request):
        return {"id": 1, "username": "test", "permissions": []}

    return dependency


auth_dependencies_stub.require_any_permission = _allow_test_user


async def _get_test_user(_request: Request):
    return {
        "id": 1,
        "username": "test",
        "permissions": ["cases.view", "cases.create", "cases.edit", "cases.delete"],
    }


auth_dependencies_stub.get_current_user = _get_test_user
sys.modules.setdefault("app.core.auth_dependencies", auth_dependencies_stub)

from app.modules.sag.backend import router as sag_router


def test_normalize_timestamp_converts_offset_to_utc():
    value = sag_router._normalize_optional_timestamp(
        "2026-07-27T12:00:00+02:00",
        "deadline",
    )

    assert value == "2026-07-27 10:00:00"


def test_normalize_timestamp_accepts_naive_datetime():
    value = sag_router._normalize_optional_timestamp(
        datetime(2026, 7, 27, 12, 30),
        "deadline",
    )

    assert value == "2026-07-27 12:30:00"


def test_normalize_timestamp_rejects_invalid_value():
    with pytest.raises(HTTPException) as exc_info:
        sag_router._normalize_optional_timestamp("not-a-date", "deadline")

    assert exc_info.value.status_code == 400


def test_attachment_path_rejects_escape_from_upload_root():
    with pytest.raises(HTTPException) as exc_info:
        sag_router._resolve_attachment_path("../../outside.txt")

    assert exc_info.value.status_code == 400


def test_attachment_path_accepts_case_subdirectory():
    path = sag_router._resolve_attachment_path("sag_files/example.txt")

    path.relative_to(sag_router.UPLOAD_BASE_PATH)


def test_upload_rejects_disallowed_extension():
    upload = UploadFile(filename="payload.html", file=BytesIO(b"<script>alert(1)</script>"))

    with pytest.raises(HTTPException) as exc_info:
        sag_router._store_upload_file(upload, sag_router.SAG_FILE_SUBDIR)

    assert exc_info.value.status_code == 400
    assert "not allowed" in str(exc_info.value.detail)


def test_relation_input_normalizes_valid_relation():
    target_id, relation_type = sag_router._normalize_relation_input(
        10,
        {"målsag_id": "11", "relationstype": " Blokkerer "},
    )

    assert target_id == 11
    assert relation_type == "Blokkerer"


@pytest.mark.parametrize(
    ("raw_type", "expected"),
    [
        ("Relateret til", "Relateret til"),
        ("Afledt af", "Afledt af"),
        ("Årsag til", "Årsag til"),
        ("afledt_af", "Afledt af"),
        ("afhænger af", "afhænger af"),
        ("undersag", "undersag"),
        ("duplikat", "duplikat"),
    ],
)
def test_relation_input_accepts_ui_and_existing_database_types(raw_type, expected):
    _, relation_type = sag_router._normalize_relation_input(
        10,
        {"målsag_id": 11, "relationstype": raw_type},
    )

    assert relation_type == expected


def test_relation_quick_task_uses_todo_steps_api():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text()

    assert "fetch(`/api/v1/sag/${caseId}/todo-steps`" in template
    assert "fetch(`/api/v1/sag/${caseId}/todos`" not in template
    assert "due_date: due" in template


def test_case_email_forward_supports_latest_mail_and_full_thread_as_new_thread():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text()

    assert "openForwardLinkedEmail('latest')" in template
    assert "openForwardLinkedEmail('thread')" in template
    assert "stripQuotedEmailHistory" in template
    assert "caseEmailComposeMode = 'forward'" in template
    assert "thread_email_id: isNewThread ? null" in template
    assert "linkedEmailsCache\n                    .filter" in template


def test_case_detail_disables_browser_cache_for_fresh_inline_ui():
    source = Path("app/modules/sag/frontend/views.py").read_text()

    assert 'response.headers["Cache-Control"] = "no-store, max-age=0"' in source


def test_case_create_lists_contacts_for_selected_customer():
    template = Path("app/modules/sag/templates/create.html").read_text()

    assert 'id="contactSearchContext"' in template
    assert "loadSelectedCustomerContacts(id);" in template
    assert "`/api/v1/customers/${customerId}/contacts`" in template
    assert "if (type === 'contact' && selectedCustomer)" in template
    assert "renderCustomerContactResults(contactInput.value);" in template
    assert "resetCustomerContactSearch();" in template


def test_case_create_defaults_responsible_to_current_user():
    template = Path("app/modules/sag/templates/create.html").read_text(encoding="utf-8")
    router = Path("app/modules/sag/backend/router.py").read_text(encoding="utf-8")
    assert "selectCurrentUserAsResponsible();" in template
    assert "raw_responsible = data.get" in router
    assert 'if "ansvarlig_bruger_id" in data else current_user_id' in router


def test_case_create_sends_relations_in_atomic_create_payload():
    template = Path("app/modules/sag/templates/create.html").read_text(encoding="utf-8")
    router = Path("app/modules/sag/backend/router.py").read_text(encoding="utf-8")
    assert "contact_ids: Object.keys(selectedContacts)" in template
    assert "telefoni_opkald_id: telefoniPrefill.callId" in template
    assert 'raw_contact_ids = data.get("contact_ids")' in router
    assert "INSERT INTO sag_kontakter" in router
    assert "UPDATE telefoni_opkald" in router
    assert "window.location.href = `/sag/${result.id}/v3`;" in template
    assert "Omdirigerer..." not in template


def test_case_can_link_internet_connection_without_customer_allocation():
    template = Path("app/modules/sag/templates/create.html").read_text(encoding="utf-8")
    sag_router = Path("app/modules/sag/backend/router.py").read_text(encoding="utf-8")
    internet_router = Path("app/modules/internet_connections/backend/router.py").read_text(encoding="utf-8")
    detail = Path("app/modules/internet_connections/templates/detail.html").read_text(encoding="utf-8")
    migration = Path("migrations/1034_sag_internet_connections.sql").read_text(encoding="utf-8")
    assert "internet_connection_ids" in template
    assert "INSERT INTO sag_internet_connections" in sag_router
    assert "/sag/{sag_id}/internet-connections" in sag_router
    assert "/internet-connections/{connection_id:int}/cases" in internet_router
    assert "Sager på forbindelsen" in detail
    assert "sag_internet_connections" in migration


def test_case_detail_has_no_overwriting_relation_functions_or_blocking_alerts():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text(encoding="utf-8")
    assert len(re.findall(r"function\s+removeContact\s*\(", template)) == 1
    assert len(re.findall(r"function\s+removeCustomer\s*\(", template)) == 1
    assert "#caseAddWorkspaceFooter .btn-primary" in template
    assert 'id="caseAddWorkspaceBody"' in template
    assert not re.search(r"(?<![\w.])alert\(", template)
    assert "reloadCasePreservingContext" in template


def test_case_list_has_direct_entity_links_and_inline_updates():
    template = Path("app/modules/sag/templates/index.html").read_text(encoding="utf-8")
    assert 'href="/customers/{{ sag.customer_id }}"' in template
    assert 'href="/contacts/{{ sag.kontakt_id }}"' in template
    assert "updateCaseListField({{ sag.id }}, 'status'" in template
    assert "updateCaseListField({{ sag.id }}, 'ansvarlig_bruger_id'" in template


def test_case_optional_data_endpoints_do_not_use_expected_404s():
    settings_router = Path("app/settings/backend/router.py").read_text(encoding="utf-8")
    subscriptions_router = Path("app/subscriptions/backend/router.py").read_text(encoding="utf-8")
    template = Path("app/modules/sag/templates/detail_v3.html").read_text(encoding="utf-8")
    assert '"time_multiplier_presets"' in settings_router
    assert '"change_request": None' in subscriptions_router
    assert "if (changePayload.change_request)" in template


def test_case_v3_contact_actions_and_company_link_include_case_context():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text(encoding="utf-8")
    assert 'href="/customers/{{ customer.id }}"' in template
    assert "sag_id: {{ case.id }}" in template
    assert "contact_id: opts.contactId || null" in template
    assert 'title="Ring til mobil"' in template
    assert 'title="Send SMS"' in template
    assert 'id="caseCallHistoryBody"' in template


def test_time_employee_picker_is_clearly_separate_from_live_tracking():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text()

    assert "Manuel registrering for" in template
    assert "Mig · vælg flere" in template
    assert "assignment_users|length" in template
    assert "#timetracking .time-v1-entry-card" in template
    assert "overflow: visible !important" in template


def test_case_email_snippet_removes_embedded_css_and_scripts():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text()

    assert "function emailSnippetAsPlainText(email)" in template
    assert "style, script, noscript, template, head" in template
    assert "const snippet = emailSnippetAsPlainText(e).slice(0, 130);" in template
    assert "snippetSource.replace(/<[^>]+>/g" not in template


def test_comment_composer_can_close_case_or_wait_for_customer():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text()

    assert 'id="commentActionCloseCase"' in template
    assert 'id="commentActionAwaitCustomer"' in template
    assert "actions.push('close_case')" in template
    assert "actions.push('await_customer')" in template
    assert "await changeCaseStatusFromComment('lukket')" in template
    assert "resolveCommentAwaitCustomerStatus()" in template
    assert "awaitCustomerCheckbox.checked = false" in template
    assert "closeCaseCheckbox.checked = false" in template


def test_comment_actions_use_styled_accessible_chips():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text()

    assert 'class="comment-action-chips"' in template
    assert template.count('class="comment-action-chip action-') == 5
    assert '.comment-action-chip:has(input:checked)' in template
    assert 'for="commentActionCloseCase"' in template
    assert 'for="commentActionAwaitCustomer"' in template


def test_case_history_shows_actor_timestamp_and_all_case_activity_filters():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text()

    assert "<strong>Hvornår:</strong>" in template
    assert "<strong>Hvem:</strong>" in template
    assert 'id="historyFilterCaseActivity"' in template
    assert "['case', 'tag', 'todo', 'relation']" in template


def test_case_timeline_uses_real_actors_and_action_timestamps():
    source = Path("app/modules/sag/backend/router.py").read_text()

    assert '"title": "Sag oprettet"' in source
    assert 'r.created_at' in source
    assert 'employee.user_id = t.medarbejder_id' in source
    assert 'LOWER(recorder.username) = LOWER(t.user_name)' in source
    assert 'Registreret for: {employee_name}' in source
    assert '"forfatter": str(row.get("actor_name") or "System")' in source
    assert '"event_type": "tag"' in source
    assert '"event_type": "todo"' in source
    assert '"event_type": "relation"' in source


def test_safe_case_html_renders_formatting_and_drops_executable_code():
    result = sanitize_safe_html(
        '<style>body{display:none}</style>'
        '<p onclick="steal()">Hej <strong>verden</strong></p>'
        '<script>alert("xss")</script>'
        '<a href="javascript:alert(1)">farligt link</a>'
    )

    assert result == (
        '<p>Hej <strong>verden</strong></p>'
        '<a target="_blank" rel="noopener noreferrer">farligt link</a>'
    )
    assert "display:none" not in result
    assert "alert" not in result
    assert "onclick" not in result


@pytest.mark.parametrize(
    "payload",
    [
        {"målsag_id": 10, "relationstype": "barn"},
        {"målsag_id": 11, "relationstype": "ukendt"},
        {"målsag_id": "ikke-et-tal", "relationstype": "barn"},
    ],
)
def test_relation_input_rejects_invalid_relation(payload):
    with pytest.raises(HTTPException) as exc_info:
        sag_router._normalize_relation_input(10, payload)

    assert exc_info.value.status_code == 400


def _request(method: str, path: str) -> Request:
    return Request(
        {
            "type": "http",
            "method": method,
            "path": path,
            "raw_path": path.encode(),
            "query_string": b"",
            "headers": [],
            "scheme": "https",
            "server": ("testserver", 443),
            "client": ("127.0.0.1", 1234),
        }
    )


def test_case_route_access_allows_view_permission_for_get():
    user = {"permissions": ["cases.view"]}

    result = asyncio.run(
        sag_router.case_route_access(_request("GET", "/api/v1/sag/10"), user)
    )

    assert result is user


def test_case_route_access_requires_edit_for_nested_post():
    user = {"permissions": ["cases.view", "cases.create"]}

    with pytest.raises(HTTPException) as exc_info:
        asyncio.run(
            sag_router.case_route_access(
                _request("POST", "/api/v1/sag/10/tags"),
                user,
            )
        )

    assert exc_info.value.status_code == 403


def test_case_sms_listener_does_not_break_click_to_call_script():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text()

    assert "console.warn('Kunne ikke logge sendt SMS i kommentarer', err);\n                }\n            });" in template
    assert "async function ringOutFromCase(number, opts = {})" in template
    assert "alert(error?.message" not in template


def test_linked_calls_and_anydesk_sessions_are_normalized_for_case_activity(monkeypatch):
    monkeypatch.setattr(sag_router, "_table_exists", lambda name: name in {"telefoni_opkald", "anydesk_sessions"})

    def fake_execute_query(query, params=None):
        if "FROM telefoni_opkald" in query:
            return [{
                "id": 11,
                "sag_id": 42,
                "direction": "outbound",
                "ekstern_nummer": "+4530480748",
                "intern_extension": "204",
                "started_at": datetime(2026, 8, 25, 10, 0),
                "ended_at": datetime(2026, 8, 25, 10, 5),
                "duration_sec": 300,
                "actor_name": "Christian",
                "contact_name": "Thomas",
            }]
        if "FROM anydesk_sessions" in query:
            return [{
                "id": 12,
                "anydesk_session_id": "ad-12",
                "sag_id": 42,
                "started_at": datetime(2026, 8, 25, 11, 0),
                "ended_at": datetime(2026, 8, 25, 11, 30),
                "duration_minutes": 30,
                "status": "completed",
                "actor_name": "Christian",
                "contact_name": "Thomas",
            }]
        return []

    monkeypatch.setattr(sag_router, "execute_query", fake_execute_query)

    activities = sag_router._get_case_linked_activities([42])

    assert [item["activity_type"] for item in activities] == ["anydesk", "call"]
    assert activities[0]["duration_minutes"] == 30
    assert activities[1]["duration_minutes"] == 5
    assert activities[1]["title"] == "Udgående opkald"


def test_time_tab_and_history_include_linked_case_activities():
    template = Path("app/modules/sag/templates/detail_v3.html").read_text()
    source = Path("app/modules/sag/backend/router.py").read_text()

    assert 'id="caseLinkedActivitiesList"' in template
    assert "/linked-activities" in template
    assert "renderCaseLinkedActivities(linkedActivities)" in template
    assert 'event_type": activity_type' in source
    assert 'source": "call" if activity_type == "call" else "anydesk"' in source


def test_sag_list_has_per_user_column_preferences():
    template = Path("app/modules/sag/templates/index.html").read_text()
    source = Path("app/modules/sag/backend/router.py").read_text()
    migration = Path("migrations/1023_user_sag_list_columns.sql").read_text()

    assert 'id="sagColumnList"' in template
    assert 'id="saveSagColumnsBtn"' in template
    assert 'draggable="true"' in template
    assert "function applySagColumnPreferences()" in template
    assert "function saveSagColumnPreferences()" in template
    assert "column_order: sagColumnOrder" in template
    assert "hidden_columns: Array.from(sagHiddenColumns)" in template
    assert "column_order: Optional[List[str]] = None" in source
    assert "hidden_columns: Optional[List[str]] = None" in source
    assert "ON CONFLICT (user_id)" in source
    assert "ADD COLUMN IF NOT EXISTS column_order JSONB" in migration
    assert "ADD COLUMN IF NOT EXISTS hidden_columns JSONB" in migration


def test_sag_list_status_dropdown_receives_options_and_links_are_styled():
    template = Path("app/modules/sag/templates/index.html").read_text()
    views = Path("app/modules/sag/frontend/views.py").read_text()

    assert '"status_options": status_options' in views
    assert "{% for status_option in status_options %}" in template
    assert 'class="sag-entity-link"' in template
    assert 'class="sag-id"' in template
    assert ".sag-entity-link:hover" in template
    assert "sag-inline-select sag-status-select" in template
    assert "sag-inline-select sag-owner-select" in template
    assert "function applyStatusSelectTone(control)" in template


def test_support_case_close_without_time_requires_explicit_confirmation():
    template = Path("app/modules/sag/templates/index.html").read_text()
    source = Path("app/modules/sag/backend/router.py").read_text()

    assert '"close_without_time_confirmation_required"' in source
    assert 'SELECT EXISTS(SELECT 1 FROM tmodule_times WHERE sag_id = %s)' in source
    assert 'confirm_close_without_time = updates.pop("confirm_close_without_time", False) is True' in source
    assert "detail?.code === 'close_without_time_confirmation_required'" in template
    assert "body.confirm_close_without_time = true" in template
    assert 'id="closeCaseWithoutTimeModal"' in template
    assert "await confirmCloseCaseWithoutTime(caseId, detail.message)" in template


def test_sag_list_has_smart_toolbar_search_and_employee_quick_filters():
    template = Path("app/modules/sag/templates/index.html").read_text()
    source = Path("app/modules/sag/backend/router.py").read_text()

    assert 'data-quick-filter="mine-open"' in template
    assert 'data-quick-filter="overdue"' in template
    assert 'data-quick-filter="my-groups"' in template
    assert 'data-quick-filter="unassigned"' in template
    assert 'id="clearSearchBtn"' in template
    assert "search.split(/\\s+/).filter(Boolean).every" in template
    assert "/sag/me/quick-filter-context" in source
    assert "SELECT group_id FROM user_groups WHERE user_id = %s" in source


def test_overdue_active_cases_have_red_row_shadow():
    template = Path("app/modules/sag/templates/index.html").read_text()

    assert ".sag-table tbody tr.sag-deadline-overdue" in template
    assert "function updateOverdueDeadlineMarker(row)" in template
    assert "!closedStatuses.has(status)" in template
    assert "row.classList.toggle('sag-deadline-overdue', isOverdue)" in template