tegwick
4153ad7305
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a0e747-8f27-7242-8df8-8bc44f88c929
60 lines
3.2 KiB
Python
60 lines
3.2 KiB
Python
"""Opt-in real User Engine HTTP source; all identities/grants are local fixtures."""
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import time
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
source = os.getenv('HUB_CORE_USER_ENGINE_SOURCE')
|
|
if not source:
|
|
pytest.skip('set HUB_CORE_USER_ENGINE_SOURCE for owner-source contract tests',allow_module_level=True)
|
|
sys.path.insert(0,str(Path(source)/'src'))
|
|
|
|
from user_engine.adapters.local import InMemoryUserEngineStore, LocalAuthorizationCheckPort
|
|
from user_engine.service import UserEngineService
|
|
from user_engine.testing.fixtures import FixtureIdentityClaimsAdapter, human_actor_claims
|
|
from user_engine.web import PortalApplication
|
|
from hub_core.security.account_reader import UserEngineAccountReader
|
|
from hub_core.security.identity import AccessFailure
|
|
from test_account_reader import body
|
|
|
|
|
|
@pytest.mark.parametrize('grant_present',[False,True])
|
|
def test_real_owner_route_serialization_and_unknown_identity(tmp_path,grant_present):
|
|
from dataclasses import replace
|
|
from user_engine.domain import PrincipalType
|
|
fixture=body(float(int(time.time())))
|
|
class Custody:
|
|
def observe(self,**kwargs): return fixture['root_grant'] if grant_present else None
|
|
class FixturePolicy(LocalAuthorizationCheckPort):
|
|
def check(self,request):
|
|
# Synthetic PDP receipt; standalone owner defaults have no decision ID.
|
|
return replace(super().check(request),decision_id='fixture-owner-decision')
|
|
store=InMemoryUserEngineStore(); store.migrate()
|
|
service=UserEngineService(store=store,identity_adapter=FixtureIdentityClaimsAdapter(),
|
|
authorization=FixturePolicy(),custody=Custody())
|
|
claims=human_actor_claims()
|
|
claims.update(iss=fixture['issuer'],sub=fixture['subject'],tenant='platform')
|
|
session=service.me(claims)
|
|
caller=replace(session.actor,subject='fixture-hub-workload',principal_type=PrincipalType.SERVICE)
|
|
token='fixture-hub-token-at-least-24-characters'
|
|
app=PortalApplication(service,trusted_proxy_secret='fixture-proxy-secret-at-least-24-characters',
|
|
login_url='https://issuer.example/login',hub_authority_caller=caller,hub_authority_token=token)
|
|
credential=tmp_path/'token'; credential.write_text(token)
|
|
before=repr(store.__dict__)
|
|
with httpx.Client(transport=httpx.WSGITransport(app=app)) as owner:
|
|
def send(request):
|
|
response=owner.post(str(request.url),headers=request.headers,content=request.content)
|
|
return httpx.Response(response.status_code,headers=response.headers,content=response.content)
|
|
async def run():
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(send),trust_env=False) as client:
|
|
reader=UserEngineAccountReader(base_url='https://owner.example',token_file=credential,client=client)
|
|
result=await reader.read(issuer=fixture['issuer'],subject=fixture['subject'],tenant='platform')
|
|
assert result.account_active and result.root_entitled == grant_present
|
|
with pytest.raises(AccessFailure):
|
|
await reader.read(issuer=fixture['issuer'],subject='unknown',tenant='platform')
|
|
asyncio.run(run())
|
|
assert repr(store.__dict__)==before
|