jobbsok/tests/test_mcp_jobbsok_tools.py

"""The `jobbsok-tools` host MCP server and its launcher (plan Step 12).

The server exists so Cowork can reach the same logic the Claude Code CLI
reaches (operator decision 7). That framing decides what is worth testing
here. The arithmetic is already covered by Steps 8 and 10; what is not covered
anywhere else is the seam -- whether the MCP layer hands back what the library
computed, whole and unaltered, or quietly reshapes it on the way out.

Three properties carry that seam, and each has a test of its own:

* **The tool surface is pinned to a golden file.** A tool list is an API. A
  renamed argument is a silent break in Cowork, where nothing type-checks the
  call, so the list is blessed once and compared byte for byte after.
* **Every tool takes an explicit `workspace`.** `jobbsok_lib.paths` refuses to
  guess at a home directory; a server that defaulted the workspace would put
  the guess back on the far side of an unsandboxed process (risk C6).
* **`avvisninger` and `advarsler` stay apart, and `vekt_hash` comes along.**
  Session 8 decided that a warning is a filter that could not run or a soft
  filter that fired, and a rejection is a hard no. Collapsing the two in the
  `tools/call` answer would erase divergences 3 and 5 at the transport layer,
  where no scoring test would ever see it.

The launcher is tested through a real subprocess in both directions, because
what it exists to prevent -- silently serving on the 3.9.6 that a GUI-spawned
process finds on an empty PATH (risk H2) -- is a property of process startup
and cannot be observed in-process. That test also covers the stdio loop's
framing, which the in-process helper deliberately does not.

Style note: this file follows tests/test_kandidatvurdering_nowrite.py.
"""

import json
import os
import subprocess
import sys

import pytest

import jobbsok_tools
from helpers import mcp_stdio

REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
SCRIPTS = os.path.join(REPO, "scripts")
GOLDEN = os.path.join(REPO, "tests", "golden", "jobbsok-tools.tools.json")
LAUNCHER = os.path.join(SCRIPTS, "jobbsok_tools_launch.py")

PROFIL_FIXTURE = ("profiles", "01-gyldig.md")
ANNONSE_FIXTURE = ("listings", "01-alt-passer.md")

#: A sub-score payload on the 0-100 contract. The corpus registers 0-10 for
#: readability, so the bridge is built here rather than in the corpus, which
#: `test_fixture_hygiene.py` holds to an exact file count.
DELSCORE = {
    "delscore": {"fagomrade": 80, "oppgavetype": 70, "teknologi": 60, "selskapstype": 50},
    "bekymringer": ["konseptfase, ikke drift"],
}


@pytest.fixture
def klient():
    """An initialized in-process MCP client bound to the server module."""
    client = mcp_stdio.Client(jobbsok_tools)
    client.initialize()
    return client


@pytest.fixture
def arbeidsomrade(empty_workspace, fixtures_dir):
    """A scaffolded workspace holding the valid fixture profile."""
    with open(os.path.join(fixtures_dir, *PROFIL_FIXTURE), "r", encoding="utf-8") as handle:
        profil = handle.read()
    with open(
        os.path.join(empty_workspace, "profil", "kandidat.md"), "w", encoding="utf-8"
    ) as handle:
        handle.write(profil)
    return empty_workspace


def les_annonse(fixtures_dir):
    from jobbsok_lib import frontmatter

    with open(os.path.join(fixtures_dir, *ANNONSE_FIXTURE), "r", encoding="utf-8") as handle:
        return frontmatter.parse(handle.read())


def test_initialize_and_tools_list_match_the_golden_surface(klient, golden):
    hilsen = klient.initialize()
    assert hilsen["protocolVersion"] == mcp_stdio.PROTOCOL_VERSION, (
        "the server must echo the client's protocol version, not choose its own"
    )
    assert hilsen["serverInfo"]["name"] == "jobbsok-tools", (
        "the server name is what .mcp.json declares; they cannot disagree"
    )
    assert "tools" in hilsen["capabilities"]

    verktoy = klient.tools()
    golden(GOLDEN, json.dumps({"tools": verktoy}, ensure_ascii=False, indent=2) + "\n")

    # The golden pins the shape; this pins the fact that M1 ships exactly three
    # tools, so Step 24 adding four more is a visible change and not a drift.
    assert [t["name"] for t in verktoy] == ["kandidat_valider", "annonse_vurder", "selvsjekk"]


def test_every_tool_requires_an_explicit_workspace(klient, arbeidsomrade):
    for verktoy in klient.tools():
        skjema = verktoy["inputSchema"]
        assert "workspace" in skjema["properties"], (
            "%s does not take a workspace argument" % verktoy["name"]
        )
        assert "workspace" in skjema.get("required", []), (
            "%s does not require workspace; there is no implicit default"
            % verktoy["name"]
        )
        # Declared as required is one thing; refused at the call is the thing
        # that matters, because Cowork does not enforce the schema for us.
        svar = klient.call(verktoy["name"], {})
        assert "error" in svar or svar["result"].get("isError") is True, (
            "%s answered a call with no workspace instead of refusing it"
            % verktoy["name"]
        )

    # And a workspace that escapes its root is refused by the same check the
    # library uses, rather than being read from wherever it resolved.
    svar = klient.call(
        "kandidat_valider", {"workspace": arbeidsomrade, "sti": "../../etc/passwd"}
    )
    assert svar["result"]["isError"] is True
    assert "workspace" in svar["result"]["content"][0]["text"].lower()


def test_the_server_and_the_library_produce_identical_output(
    klient, arbeidsomrade, fixtures_dir
):
    import kandidat_schema
    import vurdering

    fra_server = klient.call_text(
        "kandidat_valider", {"workspace": arbeidsomrade, "sti": "profil/kandidat.md"}
    )
    fra_bibliotek = kandidat_schema.report_json(
        kandidat_schema.validate_file(arbeidsomrade, "profil", "kandidat.md")
    )
    assert fra_server == fra_bibliotek, (
        "the MCP layer reshaped the validation report on its way out"
    )

    annonse, brodtekst = les_annonse(fixtures_dir)
    fra_server = klient.call_text(
        "annonse_vurder",
        {
            "workspace": arbeidsomrade,
            "annonse": annonse,
            "brodtekst": brodtekst,
            "delscore": DELSCORE["delscore"],
            "bekymringer": DELSCORE["bekymringer"],
        },
    )
    profil = vurdering.les_profil_fil(arbeidsomrade, "profil", "kandidat.md")
    fra_bibliotek = jobbsok_tools.til_json(
        vurdering.vurder(profil, annonse, brodtekst, DELSCORE)
    )
    assert fra_server == fra_bibliotek, (
        "the MCP layer reshaped the scoring result on its way out"
    )


def test_the_scoring_tool_keeps_avvisninger_advarsler_and_vekt_hash_apart(
    klient, arbeidsomrade, fixtures_dir
):
    annonse, brodtekst = les_annonse(fixtures_dir)
    # A listing that passes every hard filter but leaves salary unstated: one
    # warning, no rejection. Collapsing the two would make this read as a no.
    uten_lonn = dict(annonse)
    uten_lonn.pop("lonn_nok")

    resultat = json.loads(
        klient.call_text(
            "annonse_vurder",
            {
                "workspace": arbeidsomrade,
                "annonse": uten_lonn,
                "brodtekst": brodtekst,
                "delscore": DELSCORE["delscore"],
            },
        )
    )
    assert resultat["verdikt"] == "vurderes"
    assert resultat["avvisninger"] == []
    assert any("lonn" in a["nokkel"] for a in resultat["advarsler"]), (
        "an unstated salary must surface as a warning, not vanish: %r"
        % (resultat["advarsler"],)
    )
    assert resultat["vekt_hash"].startswith("sha256:")
    import vurdering

    assert resultat["vekt_hash"] == vurdering.vekt_hash(resultat["vekter"]), (
        "vekt_hash must be the fingerprint of the weights actually used"
    )


def test_importing_the_server_does_not_import_the_capability_modules():
    """Cold start is a budget: Cowork times an MCP server out on startup."""
    proc = subprocess.run(
        [
            sys.executable,
            "-c",
            "import sys; import jobbsok_tools; "
            "print(sorted(m for m in ('kandidat_schema', 'vurdering', "
            "'llm_ingestion_guard') if m in sys.modules))",
        ],
        cwd=REPO,
        env=dict(os.environ, PYTHONPATH=SCRIPTS),
        capture_output=True,
        text=True,
    )
    assert proc.returncode == 0, proc.stderr
    assert proc.stdout.strip() == "[]", (
        "importing the server pulled in %s at module load; the capability "
        "modules must be imported inside the handlers" % proc.stdout.strip()
    )


def test_selvsjekk_reports_the_interpreter_and_the_pinned_guard(klient, arbeidsomrade):
    rapport = json.loads(klient.call_text("selvsjekk", {"workspace": arbeidsomrade}))

    assert rapport["python_executable"] == sys.executable
    major, minor = rapport["python_version_info"][:2]
    assert (major, minor) >= (3, 10), (
        "the server is serving on Python %r; 3.10+ is the floor"
        % (rapport["python_version"],)
    )
    assert rapport["guard_versjon"] == "1.3.0", (
        "selvsjekk must report the guard version actually installed, and the "
        "pin in pyproject.toml is v1.3.0; got %r" % (rapport["guard_versjon"],)
    )
    assert rapport["workspace"] == os.path.realpath(arbeidsomrade)
    # BUILD_STAMP arrives in Step 13. Whether it is there or not, saying so
    # plainly is the contract -- a missing stamp must never read as a match.
    assert "build_stamp" in rapport


def falsk_tolk(katalog, navn, versjon="3.9.6"):
    """Write a program that fails the version gate, in the host's own form.

    The gate probes a candidate by running it, so a fake has to be a program
    and not a Python file -- a .py would be run by the real interpreter and
    would measure that one instead. On Windows that means a .cmd, which
    shutil.which resolves through PATHEXT exactly as it resolves python.exe.

    The two forms differ in one way, deliberately: cmd cannot inspect the
    probe's arguments without re-quoting a string full of parentheses and
    semicolons, so the Windows form answers unconditionally. Both fail the
    gate, which is the only thing the gate reads.

    The Windows branch is written but NOT measured -- this repository's suite
    has only ever run on macOS.
    """
    if os.name == "nt":
        sti = katalog / (navn + ".cmd")
        sti.write_text("@echo off\r\necho %s\r\nexit /b 1\r\n" % versjon)
    else:
        sti = katalog / navn
        sti.write_text(
            "#!/bin/sh\n"
            "# Answers the version probe as %s and fails the version gate.\n"
            'case "$*" in\n'
            "  *print*) echo '%s'; exit 0 ;;\n"
            "esac\n"
            "exit 1\n" % (versjon, versjon)
        )
        sti.chmod(0o755)
    return sti


def test_the_launcher_refuses_an_interpreter_below_3_10_and_serves_on_a_good_one(tmp_path):
    falsk = falsk_tolk(tmp_path, "python3.9")

    avvist = subprocess.run(
        [sys.executable, LAUNCHER],
        env=dict(os.environ, JOBBSOK_PYTHON=str(falsk)),
        input="",
        capture_output=True,
        text=True,
    )
    assert avvist.returncode != 0, (
        "the launcher started on a 3.9 interpreter instead of refusing"
    )
    assert "3.10" in avvist.stderr, (
        "the refusal must name the floor it enforced: %r" % (avvist.stderr,)
    )

    # With no candidate at all, the launcher must refuse rather than take
    # whatever python3 the PATH offers -- which under an empty launchd PATH on
    # this Mac is 3.9.6 (risk H2). Nothing in the branch above reaches this
    # path, so it needs its own case: a mutation that reinstated an ungated
    # PATH fallback survived the JOBBSOK_PYTHON case untouched.
    falsk_bin = tmp_path / "bin"
    falsk_bin.mkdir()
    falsk_tolk(falsk_bin, "python3")
    tom_rot = tmp_path / "tom-plugin-rot"
    (tom_rot / "scripts").mkdir(parents=True)

    miljo = dict(os.environ)
    miljo.pop("JOBBSOK_PYTHON", None)
    miljo.pop("CLAUDE_PLUGIN_DATA", None)
    miljo["CLAUDE_PLUGIN_ROOT"] = str(tom_rot)
    miljo["PATH"] = str(falsk_bin) + os.pathsep + os.path.dirname(sys.executable)
    uten_kandidat = subprocess.run(
        [sys.executable, LAUNCHER],
        env=miljo,
        input="",
        capture_output=True,
        text=True,
    )
    assert uten_kandidat.returncode != 0, (
        "the launcher fell through to the python3 on PATH instead of refusing"
    )
    assert "refusing to start" in uten_kandidat.stderr, (
        "the refusal must say so plainly: %r" % (uten_kandidat.stderr,)
    )

    # The other direction, over the real stdio loop: framing, flush and all.
    forespoersler = (
        json.dumps(
            {
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {"protocolVersion": mcp_stdio.PROTOCOL_VERSION},
            }
        )
        + "\n"
        + json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"})
        + "\n"
        + "\n"  # a blank line the loop must tolerate rather than answer
        + json.dumps({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
        + "\n"
    )
    servert = subprocess.run(
        [sys.executable, LAUNCHER],
        env=dict(os.environ, JOBBSOK_PYTHON=sys.executable),
        input=forespoersler,
        capture_output=True,
        text=True,
        timeout=30,
    )
    assert servert.returncode == 0, servert.stderr
    svar = [json.loads(line) for line in servert.stdout.splitlines() if line.strip()]
    assert [s["id"] for s in svar] == [1, 2], (
        "expected exactly two replies, one per request: %r" % (servert.stdout,)
    )
    assert svar[0]["result"]["serverInfo"]["name"] == "jobbsok-tools"
    assert [t["name"] for t in svar[1]["result"]["tools"]] == [
        "kandidat_valider",
        "annonse_vurder",
        "selvsjekk",
    ]