#!/usr/bin/env python3
from __future__ import annotations

import re
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]


def read(path: str) -> str:
    return (ROOT / path).read_text()


def query_sql(backend: str, name: str) -> str:
    source = read(f"src/queries/{backend}.rs")
    match = re.search(
        rf'Query::{re.escape(name)}\s*=>\s*\{{\s*r#"(.*?)"#\s*\}}',
        source,
        re.S,
    )
    if not match:
        raise AssertionError(f"{backend}: Query::{name} missing")
    return match.group(1)


def main() -> None:
    names = {
        "SHARE_LINK_INSERT": 6,
        "SHARE_LINK_PERMISSION": 3,
        "SHARE_LINK_SESSION_SOURCE": 3,
        "SHARE_SESSION_INSERT": 5,
        "SHARE_SESSION_PERMISSION": 3,
        "SHARE_SESSIONS_DELETE_BY_LINK": 3,
        "SHARE_SESSIONS_DELETE_EXPIRED": 1,
    }
    for backend in ("sqlite", "mysql"):
        for name, expected in names.items():
            sql = query_sql(backend, name)
            assert sql.count("?") == expected, (backend, name, sql)
    for name, expected in names.items():
        sql = query_sql("postgres", name)
        parameters = [int(value) for value in re.findall(r"\$(\d+)", sql)]
        assert sorted(set(parameters)) == list(range(1, expected + 1)), (name, sql)

    auth = read("src/auth/mod.rs")
    assert ".bind(&token)" not in auth[auth.index("pub async fn create_share_link"):auth.index("pub async fn update_share_link")]
    assert '"token":row.token' not in auth
    assert "token: Option<String>" not in auth[auth.index("struct SharingLinkRow"):auth.index("struct ShareLinkSessionSource")]
    assert "normalize_share_expiration" in auth
    assert "SHARE_SESSIONS_DELETE_BY_LINK" in auth
    assert "share_session_permission" in auth
    assert "invalid share session expiration in database" in auth
    assert "invalid share session permission in database" in auth
    assert "valid_share_token" in auth
    assert 'format!("share-session-client:{client_key}")' in auth
    create_link = auth[auth.index("pub async fn create_share_link"):auth.index("pub async fn update_share_link")]
    assert "no-store, max-age=0" in create_link
    create_session = auth[auth.index("pub async fn create_share_session"):auth.index("async fn share_session_permission")]
    assert "existing_session_token" in create_session
    assert 'source.permission == "ro"' in create_session
    assert '== Some("rw")' in create_session

    api = read("src/api/mod.rs")
    assert "combined_token_access_level" not in api
    request_access = api[api.index("async fn request_access_level"):api.index("fn access_level_name")]
    for required in ("share_session_token", "resource_token", "authorization_token", "account_token_access_level"):
        assert required in request_access

    access_tokens = read("src/api/access_tokens.rs")
    assert "verify_password_access_token" in access_tokens
    assert "share_access_permission" in access_tokens

    websocket = read("src/websocket/mod.rs") + read("src/websocket/pad.rs")
    assert "cookie_share_session_token" in websocket
    assert "cookie_password_token" in websocket
    assert "explicit_access_token.or(" not in websocket
    assert "verify_password_access_token" in websocket
    assert "Access expired or revoked" in websocket
    assert "access_refresh" in websocket
    assert websocket.count("update=updates.recv()=>{") == 2
    assert websocket.count('message:"Access expired or revoked"') >= 6

    pages = read("src/app/pages.rs")
    assert "canonical_resource_url" in pages
    assert "RawQuery" in pages
    assert "share_token_from_query" in pages
    assert "decode_query_component" in pages
    assert "share_session_cookie" in pages
    assert "Ok(None) => None" in pages
    assert "clear_share_session_cookie" not in pages
    assert "no-store, max-age=0" in pages
    assert "no-referrer" in pages
    assert 'decode_query_component(name).as_deref() != Some("share")' in pages

    security = read("src/security.rs")
    assert "__Host-rustpad_share_" in security
    assert "HttpOnly; Secure; SameSite=Lax" in security

    for backend in ("sqlite", "postgres", "mysql"):
        sharing_list = query_sql(backend, "RESOURCE_SHARING_LINKS")
        assert not re.search(r"(?:^|,)\s*(?:CAST\()?token\b", sharing_list)
        migration = read(f"migrations/{backend}/0026_share_link_sessions.sql")
        assert "UPDATE resource_share_links SET token = NULL" in migration
        assert "session_token_hash" in migration
        assert "ON DELETE CASCADE" in migration

    workspace_js = read("static/js/workspace.js")
    note_js = read("static/js/note-editor.js")
    assert 'info.access_level === "none"' in workspace_js
    assert note_js.count('info.access_level === "none"') >= 2

    app = read("src/app/mod.rs")
    assert "PathOnlyMakeSpan" in app
    assert "request.uri().path()" in app
    assert "TraceLayer::new_for_http().make_span_with(PathOnlyMakeSpan)" in app

    print("share session static regression tests: passed")


if __name__ == "__main__":
    main()