mail-address-semantics/tests/test_discovery.py

import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import pytest

from mas_conformance import statute_document
from mas_core.discovery import (
    DiscoveryCache,
    FailureKind,
    Response,
    discover,
)

GS = statute_document("valid-great-service.json")
JSON = {"content-type": "application/json"}


def fake(routes):
    calls = []

    def transport(url, timeout):
        calls.append(url)
        r = routes.get(url)
        if r is None:
            return Response(404, {}, b"")
        if isinstance(r, Exception):
            raise r
        return r
    transport.calls = calls
    return transport


def ok(doc, ctype="application/json"):
    return Response(200, {"content-type": ctype}, json.dumps(doc).encode())


def redirect(loc, status=301):
    return Response(status, {"location": loc}, b"")


U = "https://great-service.example/.well-known/mas"


def test_success_exact_host():
    r = discover("great-service.example", transport=fake({U: ok(GS)}))
    assert r.ok and r.statute.namespace == "great-service.example"


def test_no_parent_walk():
    t = fake({U: ok(GS)})
    r = discover("acme.great-service.example", transport=t)
    assert not r.ok and r.kind is FailureKind.NotFound
    assert t.calls == ["https://acme.great-service.example/.well-known/mas"]


@pytest.mark.parametrize("routes,kind", [
    ({U: Response(500, {}, b"")}, FailureKind.HttpFailure),
    ({U: OSError("connection refused")}, FailureKind.TransportFailure),
    ({U: ok(GS, "text/html")}, FailureKind.InvalidContentType),
    ({U: Response(200, JSON, b"{not json")}, FailureKind.NotJson),
    ({U: ok({**GS, "mas_version": "0.2"})}, FailureKind.UnsupportedVersion),
    ({U: ok({**GS, "profiles": {"p1": True}})}, FailureKind.SchemaInvalid),
    ({U: ok(statute_document("semantic-p2-disabled-modes.json"))}, FailureKind.SemanticInvalid),
    ({U: ok({**GS, "namespace": "other.example"})}, FailureKind.NamespaceMismatch),
    ({U: redirect("https://evil.example/.well-known/mas")}, FailureKind.CrossHostRedirect),
    ({U: redirect("https://great-service.example:8443/mas")}, FailureKind.CrossHostRedirect),
    ({U: redirect("http://great-service.example/.well-known/mas")}, FailureKind.SchemeDowngrade),
    ({U: redirect("/.well-known/mas")}, FailureKind.TooManyRedirects),
])
def test_typed_failures(routes, kind):
    r = discover("great-service.example", transport=fake(routes))
    assert not r.ok and r.kind is kind, r


def test_same_origin_redirect_followed():
    alt = "https://great-service.example/static/mas.json"
    r = discover("great-service.example", transport=fake({U: redirect("/static/mas.json", 302), alt: ok(GS)}))
    assert r.ok


def _child(parent_url, **over):
    return {**GS, "inherits": parent_url, **over}


def test_inheritance_via_discovery():
    P = "https://parent.example/.well-known/mas"
    parent = {**GS, "namespace": "parent.example",
              "resource_types": {"invoice": {"enabled": True}}, "addresses": {"boss": {"kind": "person"}}}
    r = discover("great-service.example", transport=fake({U: ok(_child(P)), P: ok(parent)}))
    assert r.ok and r.statute.resource_type_enabled("invoice") and r.statute.resource_type_enabled("ticket")
    assert "boss" not in r.statute.addresses


def test_inheritance_cycle_via_discovery():
    P = "https://parent.example/.well-known/mas"
    parent = {**GS, "namespace": "parent.example", "inherits": U}
    r = discover("great-service.example", transport=fake({U: ok(_child(P)), P: ok(parent)}))
    assert not r.ok and r.kind is FailureKind.InheritanceCycle


def test_inheritance_parent_failure_is_typed():
    P = "https://parent.example/.well-known/mas"
    r = discover("great-service.example", transport=fake({U: ok(_child(P))}))
    assert not r.ok and r.kind is FailureKind.InheritanceFailure and "NotFound" in r.detail


def test_cache_keeps_failures_typed_and_separate():
    now = [0.0]
    cache = DiscoveryCache(success_ttl=100, failure_ttl=10, clock=lambda: now[0])
    routes = {}
    t = fake(routes)
    r1 = discover("great-service.example", transport=t, cache=cache)
    r2 = discover("great-service.example", transport=t, cache=cache)
    assert r1.kind is r2.kind is FailureKind.NotFound and r2.cached and len(t.calls) == 1
    routes[U] = ok(GS)
    now[0] = 11
    r3 = discover("great-service.example", transport=t, cache=cache)
    assert r3.ok and not r3.cached
    now[0] = 50
    assert discover("great-service.example", transport=t, cache=cache).cached


class _Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/.well-known/mas":
            body = json.dumps({**GS, "namespace": "localhost"}).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
        else:
            self.send_response(404)
            self.end_headers()

    def log_message(self, *a):
        pass


def test_local_http_fixture_server():
    srv = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    try:
        r = discover("localhost", scheme="http", port=srv.server_address[1])
        assert r.ok and r.statute.namespace == "localhost"
        r = discover("LOCALHOST.", scheme="http", port=srv.server_address[1])
        assert r.ok
    finally:
        srv.shutdown()