Commit d0db107e authored by Sergio Gimenez's avatar Sergio Gimenez
Browse files

fix tests

parent b479cbc4

tests/integration/test_partner_repo.py

+22 −12
Original line number Diff line number Diff line
@@ -13,11 +13,20 @@ from federation_manager.adapters.database.partner_repo import PostgresPartnerRep 
from federation_manager.adapters.database.tables import partner_ops

from federation_manager.application.authentication import PartnerAuthenticator

from federation_manager.domain.errors import PartnerNotActive

from federation_manager.domain.models import ValidatedClaims

from federation_manager.domain.ports import PartnerRepositoryPort

from tests.fakes import FakeJwtValidator



pytestmark = pytest.mark.integration



URL = os.getenv("FM_POSTGRES_URL", "postgresql+asyncpg://fm:fm@localhost:5433/fm_db")

TOKEN = "token-partner-a"





def _validator(client_id: str) -> FakeJwtValidator:

    return FakeJwtValidator(

        {TOKEN: ValidatedClaims(client_id=client_id, scopes={"fed-mgmt"})}

    )





async def test_repo_reads_partner_from_postgres() -> None:
@@ -25,27 +34,28 @@ async def test_repo_reads_partner_from_postgres() -> None: 
    await create_schema(engine)

    session_maker = build_session_maker(engine)



    thumbprint = f"AA:{uuid4().hex[:8]}"

    client_id = f"partner-{uuid4().hex[:8]}"

    async with session_maker() as session:

        await session.execute(

            insert(partner_ops).values(

                id=uuid4(), mcc_mnc=uuid4().hex[:10], cert_thumbprint=thumbprint, status="active"

                id=uuid4(), mcc_mnc=uuid4().hex[:10], oauth2_client_id=client_id, status="active"

            )

        )

        await session.commit()



        repo: PartnerRepositoryPort = PostgresPartnerRepo(session)

        partner = await repo.find_by_cert_thumbprint(thumbprint)

        partner = await repo.find_by_oauth2_client_id(client_id)



        assert partner is not None

        assert partner.cert_thumbprint == thumbprint

        assert partner.oauth2_client_id == client_id

        assert partner.is_active()



        assert await PartnerAuthenticator(repo).authenticate(thumbprint) is not None

        assert await repo.find_by_cert_thumbprint("nope") is None

        auth = PartnerAuthenticator(repo, _validator(client_id))

        assert await auth.authenticate(TOKEN) is not None

        assert await repo.find_by_oauth2_client_id("nope") is None



        await session.execute(

            delete(partner_ops).where(partner_ops.c.cert_thumbprint == thumbprint)

            delete(partner_ops).where(partner_ops.c.oauth2_client_id == client_id)

        )

        await session.commit()
@@ -57,24 +67,24 @@ async def test_suspended_partner_rejected_against_postgres() -> None: 
    await create_schema(engine)

    session_maker = build_session_maker(engine)



    thumbprint = f"BB:{uuid4().hex[:8]}"

    client_id = f"partner-{uuid4().hex[:8]}"

    async with session_maker() as session:

        await session.execute(

            insert(partner_ops).values(

                id=uuid4(),

                mcc_mnc=uuid4().hex[:10],

                cert_thumbprint=thumbprint,

                oauth2_client_id=client_id,

                status="suspended",

            )

        )

        await session.commit()



        auth = PartnerAuthenticator(PostgresPartnerRepo(session))

        auth = PartnerAuthenticator(PostgresPartnerRepo(session), _validator(client_id))

        with pytest.raises(PartnerNotActive):

            await auth.authenticate(thumbprint)

            await auth.authenticate(TOKEN)



        await session.execute(

            delete(partner_ops).where(partner_ops.c.cert_thumbprint == thumbprint)

            delete(partner_ops).where(partner_ops.c.oauth2_client_id == client_id)

        )

        await session.commit()

tests/test_authentication.py

+41 −11
Original line number Diff line number Diff line
@@ -3,31 +3,61 @@ from uuid import uuid4 
import pytest



from federation_manager.application.authentication import PartnerAuthenticator

from federation_manager.domain.errors import PartnerNotActive, PartnerUnknown

from federation_manager.domain.models import PartnerOP

from tests.fakes import InMemoryPartnerRepo

from federation_manager.domain.errors import (

    AuthenticationFailed,

    PartnerNotActive,

    PartnerUnknown,

)

from federation_manager.domain.models import PartnerOP, ValidatedClaims

from tests.fakes import FakeJwtValidator, InMemoryPartnerRepo



CLIENT_ID = "partner-a"

TOKEN = "token-partner-a"





def _partner(status: str = "active") -> PartnerOP:

    return PartnerOP(id=uuid4(), mcc_mnc="214-07", cert_thumbprint="AA:BB", status=status)

    return PartnerOP(id=uuid4(), mcc_mnc="214-07", oauth2_client_id=CLIENT_ID, status=status)





def _authenticator(partner: PartnerOP, scopes: set[str] | None = None) -> PartnerAuthenticator:

    claims = ValidatedClaims(client_id=CLIENT_ID, scopes=scopes or {"fed-mgmt"})

    return PartnerAuthenticator(

        InMemoryPartnerRepo([partner]), FakeJwtValidator({TOKEN: claims})

    )





async def test_authenticates_known_active_partner() -> None:

    partner = _partner()

    auth = PartnerAuthenticator(InMemoryPartnerRepo([partner]))



    assert await auth.authenticate("AA:BB") is partner

    assert await _authenticator(partner).authenticate(TOKEN) is partner





async def test_invalid_token_is_rejected() -> None:

    auth = _authenticator(_partner())



    with pytest.raises(AuthenticationFailed):

        await auth.authenticate("not-a-real-token")





async def test_token_without_fed_mgmt_scope_is_rejected() -> None:

    auth = _authenticator(_partner(), scopes={"some-other-scope"})



    with pytest.raises(AuthenticationFailed):

        await auth.authenticate(TOKEN)





async def test_unknown_thumbprint_is_rejected() -> None:

    auth = PartnerAuthenticator(InMemoryPartnerRepo([_partner()]))

async def test_unknown_client_id_is_rejected() -> None:

    claims = ValidatedClaims(client_id="partner-nobody", scopes={"fed-mgmt"})

    auth = PartnerAuthenticator(

        InMemoryPartnerRepo([_partner()]), FakeJwtValidator({TOKEN: claims})

    )



    with pytest.raises(PartnerUnknown):

        await auth.authenticate("CC:DD")

        await auth.authenticate(TOKEN)





async def test_suspended_partner_is_rejected() -> None:

    auth = PartnerAuthenticator(InMemoryPartnerRepo([_partner(status="suspended")]))

    auth = _authenticator(_partner(status="suspended"))



    with pytest.raises(PartnerNotActive):

        await auth.authenticate("AA:BB")
 
        await auth.authenticate(TOKEN)

tests/test_authorization.py

+20 −14
Original line number Diff line number Diff line
@@ -6,15 +6,16 @@ import pytest 
from federation_manager.application.authentication import PartnerAuthenticator

from federation_manager.application.authorization import FederationAuthorizer

from federation_manager.domain.errors import AgreementExpired, AgreementViolation

from federation_manager.domain.models import Agreement, PartnerOP

from tests.fakes import InMemoryAgreementRepo, InMemoryPartnerRepo

from federation_manager.domain.models import Agreement, PartnerOP, ValidatedClaims

from tests.fakes import FakeJwtValidator, InMemoryAgreementRepo, InMemoryPartnerRepo



THUMBPRINT = "AA:BB"

CLIENT_ID = "partner-a"

TOKEN = "token-partner-a"

APP_ID = "partner-app"





def _partner() -> PartnerOP:

    return PartnerOP(id=uuid4(), mcc_mnc="214-07", cert_thumbprint=THUMBPRINT, status="active")

    return PartnerOP(id=uuid4(), mcc_mnc="214-07", oauth2_client_id=CLIENT_ID, status="active")





def _agreement(partner_id: object, **over: object) -> Agreement:
@@ -33,9 +34,15 @@ def _agreement(partner_id: object, **over: object) -> Agreement: 
    return Agreement(**base)  # type: ignore[arg-type]





def _authenticator(partner: PartnerOP) -> PartnerAuthenticator:

    claims = ValidatedClaims(client_id=CLIENT_ID, scopes={"fed-mgmt"})

    return PartnerAuthenticator(

        InMemoryPartnerRepo([partner]), FakeJwtValidator({TOKEN: claims})

    )





def _authorizer(partner: PartnerOP, agreement: Agreement) -> FederationAuthorizer:

    auth = PartnerAuthenticator(InMemoryPartnerRepo([partner]))

    return FederationAuthorizer(auth, InMemoryAgreementRepo([agreement]))

    return FederationAuthorizer(_authenticator(partner), InMemoryAgreementRepo([agreement]))





async def test_authorizes_valid_request_and_resolves_spec() -> None:
@@ -47,7 +54,7 @@ async def test_authorizes_valid_request_and_resolves_spec() -> None: 
    )



    result = await _authorizer(partner, agreement).authorize(

        THUMBPRINT, "edge-cloud-deploy", APP_ID, zone_id=zone

        TOKEN, "edge-cloud-deploy", APP_ID, zone_id=zone

    )



    assert result.partner is partner
@@ -59,16 +66,15 @@ async def test_expired_agreement_rejected() -> None: 
    agreement = _agreement(partner.id, valid_until=datetime(2025, 1, 1, tzinfo=timezone.utc))



    with pytest.raises(AgreementExpired):

        await _authorizer(partner, agreement).authorize(THUMBPRINT, "edge-cloud-deploy", APP_ID)

        await _authorizer(partner, agreement).authorize(TOKEN, "edge-cloud-deploy", APP_ID)





async def test_no_active_agreement_rejected() -> None:

    partner = _partner()

    auth = PartnerAuthenticator(InMemoryPartnerRepo([partner]))

    authorizer = FederationAuthorizer(auth, InMemoryAgreementRepo([]))

    authorizer = FederationAuthorizer(_authenticator(partner), InMemoryAgreementRepo([]))



    with pytest.raises(AgreementExpired):

        await authorizer.authorize(THUMBPRINT, "edge-cloud-deploy", APP_ID)

        await authorizer.authorize(TOKEN, "edge-cloud-deploy", APP_ID)





async def test_api_not_permitted_rejected() -> None:
@@ -76,7 +82,7 @@ async def test_api_not_permitted_rejected() -> None: 
    agreement = _agreement(partner.id, permitted_api_types={"device-location"})



    with pytest.raises(AgreementViolation):

        await _authorizer(partner, agreement).authorize(THUMBPRINT, "edge-cloud-deploy", APP_ID)

        await _authorizer(partner, agreement).authorize(TOKEN, "edge-cloud-deploy", APP_ID)





async def test_zone_not_permitted_rejected() -> None:
@@ -85,7 +91,7 @@ async def test_zone_not_permitted_rejected() -> None: 


    with pytest.raises(AgreementViolation):

        await _authorizer(partner, agreement).authorize(

            THUMBPRINT, "edge-cloud-deploy", APP_ID, zone_id=uuid4()

            TOKEN, "edge-cloud-deploy", APP_ID, zone_id=uuid4()

        )
@@ -94,4 +100,4 @@ async def test_unmapped_app_rejected() -> None: 
    agreement = _agreement(partner.id, service_spec_mappings={"other-app": uuid4()})



    with pytest.raises(AgreementViolation):

        await _authorizer(partner, agreement).authorize(THUMBPRINT, "edge-cloud-deploy", APP_ID)
 
        await _authorizer(partner, agreement).authorize(TOKEN, "edge-cloud-deploy", APP_ID)

tests/test_ewbi_management.py

+40 −16
Original line number Diff line number Diff line
@@ -5,13 +5,15 @@ from uuid import uuid4 
from fastapi import FastAPI

from fastapi.testclient import TestClient



from federation_manager.dependencies import get_partner_repo

from federation_manager.domain.models import PartnerOP

from federation_manager.dependencies import get_jwt_validator, get_partner_repo

from federation_manager.domain.models import PartnerOP, ValidatedClaims

from federation_manager.main import create_app

from tests.fakes import InMemoryPartnerRepo

from tests.fakes import FakeJwtValidator, InMemoryPartnerRepo



THUMBPRINT = "AA:BB"

HEADER = "X-Client-Certificate-Thumbprint"

CLIENT_ID = "partner-a"

TOKEN = "token-partner-a"

STRANGER_TOKEN = "token-partner-nobody"

URL = "/ewbi/v1/management/heartbeat"





@asynccontextmanager
@@ -19,42 +21,64 @@ async def _no_infra(app: FastAPI) -> AsyncIterator[None]: 
    yield





def _bearer(token: str) -> dict[str, str]:

    return {"Authorization": f"Bearer {token}"}





def _client(status: str = "active") -> TestClient:

    partner = PartnerOP(id=uuid4(), mcc_mnc="214-07", cert_thumbprint=THUMBPRINT, status=status)

    partner = PartnerOP(id=uuid4(), mcc_mnc="214-07", oauth2_client_id=CLIENT_ID, status=status)

    validator = FakeJwtValidator(

        {

            TOKEN: ValidatedClaims(client_id=CLIENT_ID, scopes={"fed-mgmt"}),

            STRANGER_TOKEN: ValidatedClaims(client_id="partner-nobody", scopes={"fed-mgmt"}),

        }

    )

    app = create_app(lifespan=_no_infra)

    app.dependency_overrides[get_partner_repo] = lambda: InMemoryPartnerRepo([partner])

    app.dependency_overrides[get_jwt_validator] = lambda: validator

    return TestClient(app)





def test_heartbeat_accepts_active_partner() -> None:

    r = _client().post("/ewbi/v1/management/heartbeat", headers={HEADER: THUMBPRINT})

    r = _client().post(URL, headers=_bearer(TOKEN))

    assert r.status_code == 200

    assert r.json() == {"status": "ALIVE"}





def test_missing_certificate_header_is_401() -> None:

    r = _client().post("/ewbi/v1/management/heartbeat")

def test_missing_authorization_header_is_401() -> None:

    r = _client().post(URL)

    assert r.status_code == 401

    assert r.json()["type"] == "urn:oop:ewbi:error:authentication-failed"

    assert r.headers["WWW-Authenticate"] == 'Bearer scope="fed-mgmt"'





def test_non_bearer_scheme_is_401() -> None:

    r = _client().post(URL, headers={"Authorization": "Basic partner-a:secret"})

    assert r.status_code == 401

    assert r.json()["type"] == "urn:oop:ewbi:error:authentication-failed"





def test_unverifiable_token_is_401() -> None:

    r = _client().post(URL, headers=_bearer("forged"))

    assert r.status_code == 401

    assert r.json()["type"] == "urn:oop:ewbi:error:authentication-failed"





def test_unknown_partner_is_401() -> None:

    r = _client().post("/ewbi/v1/management/heartbeat", headers={HEADER: "ZZ:ZZ"})

    r = _client().post(URL, headers=_bearer(STRANGER_TOKEN))

    assert r.status_code == 401

    body = r.json()

    assert body["type"] == "urn:oop:ewbi:error:partner-unknown"

    assert body["instance"] == "/ewbi/v1/management/heartbeat"

    assert body["instance"] == URL





def test_suspended_partner_is_403() -> None:

    r = _client(status="suspended").post(

        "/ewbi/v1/management/heartbeat", headers={HEADER: THUMBPRINT}

    )

    r = _client(status="suspended").post(URL, headers=_bearer(TOKEN))

    assert r.status_code == 403

    assert r.json()["type"] == "urn:oop:ewbi:error:partner-not-active"





def test_error_response_leaks_no_internals() -> None:

    r = _client().post("/ewbi/v1/management/heartbeat", headers={HEADER: "ZZ:ZZ"})

    assert "ZZ:ZZ" not in r.text

    r = _client().post(URL, headers=_bearer(STRANGER_TOKEN))

    assert "partner-nobody" not in r.text

    assert "partner_ops" not in r.text

tests/test_ports.py

+5 −3
Original line number Diff line number Diff line
@@ -6,8 +6,10 @@ from tests.fakes import InMemoryPartnerRepo 




async def test_fake_satisfies_port_and_finds_partner() -> None:

    partner = PartnerOP(id=uuid4(), mcc_mnc="214-07", cert_thumbprint="AA:BB", status="active")

    partner = PartnerOP(

        id=uuid4(), mcc_mnc="214-07", oauth2_client_id="partner-a", status="active"

    )

    repo: PartnerRepositoryPort = InMemoryPartnerRepo([partner])



    assert await repo.find_by_cert_thumbprint("AA:BB") is partner

    assert await repo.find_by_cert_thumbprint("unknown") is None
 
    assert await repo.find_by_oauth2_client_id("partner-a") is partner

    assert await repo.find_by_oauth2_client_id("unknown") is None