"""Origin + two real TLS providers: explicit fresh-agreement handover. No copied operational store, secret, token, mandate or archive grants access. Uses the existing closed wire contracts; this procedure adds no wire profile. """ import base64 import copy import http.client import http.server import ssl import threading import unittest from types import SimpleNamespace from unittest import mock from urllib.parse import urlsplit from cryptography.hazmat.primitives.asymmetric import ec from odexa_ref import crypto from profiles import delegation as d, portable_evidence as p, paid_evidence as paid from profiles import storage_evidence as storage from profiles import paid_contracts as pc from profiles.authority_history import AuthorityHistory from profiles.authority_transport import ScopedCredential from profiles.free_gateway import FreeGateway from profiles.free_service import FreeAgreementService, uid, later from profiles.free_store import FreeStore from profiles.paid_service import PaidAgreementService from profiles.paid_store import PaidStore, TransactionPayments import test_free_service as fixture import test_paid_evidence as paid_fixture from test_storage_evidence import enable_export from test_storage_service import configure, agree, action, session, signed class ProviderHandoverTests(unittest.TestCase): @classmethod def setUpClass(cls): fixture.FreeServiceTests.setUpClass() ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) root = fixture.FreeServiceTests.tls_path ctx.load_cert_chain(root/'tls-cert.pem', root/'tls-key.pem') cls.successor_server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), fixture.Handler) cls.successor_server.daemon_threads = True cls.successor_server.socket = ctx.wrap_socket(cls.successor_server.socket, server_side=True) cls.thread = threading.Thread(target=cls.successor_server.serve_forever, daemon=True) cls.thread.start() cls.successor_origin = f'https://127.0.0.1:{cls.successor_server.server_port}' @classmethod def tearDownClass(cls): cls.successor_server.shutdown(); cls.successor_server.server_close(); cls.thread.join(timeout=2) fixture.FreeServiceTests.tearDownClass() def setUp(self): self.f = fixture.FreeServiceTests(); self.f.setUp(); enable_export(self.f) self.successor_server.requests = [] self.clock_patch = mock.patch('profiles.authority_transport.utc_now', side_effect=lambda: self.f.now()) self.clock_patch.start() self.asset = b'Origin-controlled handover test article.\n' self.finished = threading.Event() self.delivery_error = None self.selected_gateway = None self.f.origin_server.route = self.origin_route def tearDown(self): self.clock_patch.stop(); self.f.tearDown() def http(self, host, path, *, body=None, auth=None, method='POST', media='application/json', expected=200): """Verified HTTPS; explicit negative auth cases may receive non-2xx.""" url = urlsplit(host) conn = http.client.HTTPSConnection(url.hostname, url.port, timeout=5, context=ssl.create_default_context(cafile=str(self.f.tls_path/'ca.pem'))) headers = {'Content-Type': media} if auth is not None: headers['Authorization'] = auth raw = body if isinstance(body, bytes) else crypto.json_bytes(body) if body is not None else b'' try: conn.request(method, path, body=raw, headers=headers) response = conn.getresponse(); result = response.read() self.assertEqual(response.status, expected, result) return result finally: conn.close() def origin_route(self, method, path, headers, body): f = self.f if method != 'GET' or path != '/about': return f.origin_route(method, path, headers, body) try: authorization = headers.get('Authorization', '') if self.selected_gateway is None or not authorization.startswith('Bearer '): raise ValueError('no access') gateway = self.selected_gateway rid = gateway.admit(authorization[7:], resource_url=f.request['url'], actions=['retrieve'], purposes=['public_retrieval']) except ValueError: return 403, {'Content-Type': 'application/json'}, b'{}' def finish(count, complete): try: gateway.finish_delivery(rid, bytes_written=count, complete=complete, representation_digest=crypto.digest(self.asset)) except Exception as error: self.delivery_error = error finally: self.finished.set() return 200, {'Content-Type': 'text/plain'}, self.asset, { 'begin': lambda: gateway.begin_delivery(rid), 'finish': finish} def resource(self, token, *, expected=200): self.finished.clear(); self.delivery_error = None raw = self.http(self.f.origin, '/about', method='GET', auth='Bearer '+token, expected=expected) if expected == 200: self.assertTrue(self.finished.wait(5)) if self.delivery_error: raise self.delivery_error self.assertEqual(raw, self.asset) return raw def gateway(self, who, name): return FreeGateway(self.f.path/(name+'.sqlite3'), binding=who.binding, base_url=who.base, transport=self.f.transport, history=AuthorityHistory(self.f.path/(name+'-history.sqlite3')), credential=ScopedCredential(who.host, urlsplit(who.base).path, who.clients['gateway']['authorization']), clock=self.f.now) def prepare(self, *, paid_fixture=None): f = self.f; host = self.successor_origin b = SimpleNamespace(host=host, base=host+'/successor/api/', principal=f.principal) b.signing_key = ec.derive_private_key(71, ec.SECP256R1()) b.agent_key = ec.derive_private_key(72, ec.SECP256R1()) b.gateway_key = ec.derive_private_key(73, ec.SECP256R1()) b.descriptor = copy.deepcopy(f.authority['services'][0]) b.descriptor.update(id=f.origin+'/services/successor', base_url=b.base, issuer=host+'/operator') b.descriptor['signing_keys'] = [dict(b.descriptor['signing_keys'][0], kid=host+'/keys/one', public_jwk=crypto.public_jwk(b.signing_key))] b.descriptor['limits']['payment_mode'] = 'external' if paid_fixture else 'none' b.grant = copy.deepcopy(f.authority['delegations'][0]) b.grant.update(id=f.origin+'/delegations/successor', delegate_service_id=b.descriptor['id'], payment_mode=b.descriptor['limits']['payment_mode']) b.binding = dict(f.binding, service_id=b.descriptor['id'], issuer=b.descriptor['issuer'], delegation_id=b.grant['id']) identity = {k:b.binding[k] for k in ('origin','service_id','issuer')} b.store = (PaidStore if paid_fixture else FreeStore)(f.path/'successor.sqlite3', identity) b.history = AuthorityHistory(f.path/'successor-history.sqlite3') b.args = dict(binding=b.binding, base_url=b.base, key_id=b.descriptor['signing_keys'][0]['kid'], private_key=b.signing_key, terms_bytes=b'Successor terms: fresh explicit agreement.\n', transport=f.transport, history=b.history, clock=f.now) b.extra_services = []; b.extra_grants = [] if paid_fixture: self.configure_paid_successor(b, paid_fixture) else: b.service = FreeAgreementService(b.store, **b.args) b.clients = {} for index, (role, key) in enumerate([('agent',b.agent_key),('gateway',b.gateway_key),('admin',None)]): secret = chr(120+index)*43 descriptor = b.service.provision_client(role, secret, [f.principal], role=role, public_jwk=crypto.public_jwk(key) if key else None) descriptor['authorization'] = 'Basic '+base64.b64encode((role+':'+secret).encode()).decode() b.clients[role] = descriptor self.successor_server.route = b.route if paid_fixture else b.service.dispatch self.b = b return b def configure_paid_successor(self, b, old): f = self.f b.payment_key = ec.derive_private_key(74, ec.SECP256R1()) b.payer_key = ec.derive_private_key(75, ec.SECP256R1()); b.payer_id = old.quote_payer b.payer_kid = b.host+'/payer/key'; b.payment_kid = b.host+'/payment/key' b.verifier = copy.deepcopy(old.verifier) b.verifier.update(id=f.origin+'/services/successor-payment', base_url=b.host+'/payment/', issuer=b.host+'/payment/operator') b.verifier['signing_keys'] = [dict(b.verifier['signing_keys'][0], kid=b.payment_kid, public_jwk=crypto.public_jwk(b.payment_key))] b.verifier_grant = dict(copy.deepcopy(old.verifier_grant), id=f.origin+'/delegations/successor-payment', delegate_service_id=b.verifier['id']) b.extra_services = [b.verifier]; b.extra_grants = [b.verifier_grant] b.verify_credential = ScopedCredential(b.host, '/payment/', 'Bearer successor-verifier-only') def quote(offer, payer_id): value = crypto.strict_json(old.quote(offer, payer_id)) value.update(service_id=b.binding['service_id'], verification_service_id=b.verifier['id'], provider_id=b.verifier['issuer'], verification_endpoint=b.verifier['base_url']+'verify') value['items'][1]['payee_id'] = b.host+'/payee' return crypto.json_bytes(value) b.service = PaidAgreementService(b.store, **b.args, quote_factory=quote, verifier_delegation_id=b.verifier_grant['id'], verifier_credential=b.verify_credential) b.service.provision_payer('payer', 's'*43, b.payer_id, key_id=b.payer_kid, public_jwk=crypto.public_jwk(b.payer_key)) b.payer_auth = 'Basic '+base64.b64encode(b'payer:'+b's'*43).decode(); b.verify_calls = 0 b.responses = {} def route(method, path, headers, body): if path != '/payment/verify': return b.service.dispatch(method,path,headers,body) b.verify_calls += 1 if method != 'POST' or headers.get('Authorization') != b.verify_credential.authorization: return 401, {'Content-Type':'application/json'}, b'{}' check = crypto.strict_json(body) if check['check_id'] not in b.responses: value = dict(check, record_type='payment.verification', sequence=len(b.responses)+1, state='confirmed', provider_reference='synthetic-successor-check', effective_at=f.now(), checked_at=f.now()) b.responses[check['check_id']] = crypto.sign_jws(crypto.json_bytes(value), b.payment_key, b.payment_kid, 'odexa-payment-status+jws', allowed_types={'odexa-payment-status+jws'}).encode() return 200, {'Content-Type':'application/jose'}, b.responses[check['check_id']] b.route = route def call_b(self, route, value=None, *, role='agent', auth=None, expected=200, method='POST'): b = self.b return self.http(b.host, urlsplit(b.base).path+route, body=value, method=method, auth=auth if auth is not None else b.clients[role]['authorization'], expected=expected, media='application/jose' if isinstance(value,bytes) else 'application/json') def publish(self, *, withdraw=False): f, b = self.f, self.b if not any(s['id'] == b.descriptor['id'] for s in f.authority['services']): f.authority['services'].extend([b.descriptor, *b.extra_services]) f.authority['delegations'].extend([b.grant, *b.extra_grants]) if withdraw: f.authority['delegations'] = [g for g in f.authority['delegations'] if g['delegate_service_id'] in {b.descriptor['id'], *(s['id'] for s in b.extra_services)}] for service in f.authority['services']: if service['id'] not in {b.descriptor['id'], *(s['id'] for s in b.extra_services)}: for key in service['signing_keys']: key.update(state='retired', retired_at=f.now()) f.authority['revision'] += 1 d.validate_authority(f.authority, f.policy) def agree_b(self, *, paid_mode=False): f, b = self.f, self.b request = dict(b.binding, principal_id=b.principal, request=f.request, access_seconds=120, use_seconds=240) if paid_mode: request['payer_id'] = b.payer_id b.offer_raw = self.call_b('offers', request, expected=201); b.offer = crypto.strict_json(b.offer_raw) assent = dict(b.binding, offer_id=b.offer['offer_id'], offer_digest=crypto.digest(b.offer_raw), client_id='agent', principal_id=b.principal, intent='accept', nonce=b.offer['nonce'], accepted_at=f.now(), idempotency_key='successor-acceptance-1') b.assent = crypto.sign_jws(crypto.json_bytes(assent),b.agent_key,b.clients['agent']['key_id'],'odexa-acceptance+jws').encode() b.receipt_jws = self.call_b('agreements',b.assent,expected=201) b.receipt = crypto.strict_json(crypto.verify_jws(b.receipt_jws,crypto.public_jwk(b.signing_key),'odexa-receipt+jws')[0]) b.agreement_id = b.receipt['agreement_id'] return b.receipt def token_b(self, *, expected=200): raw = self.call_b('tokens',dict(self.b.binding,agreement_id=self.b.agreement_id),expected=expected) return crypto.strict_json(raw).get('access_token') def export_old(self, *, selected_storage=False, who=None): f = self.f; source = who or f; snapshot = f.transport.fetch(f.origin) operation = dict(operation_id=uid(),payment_request_digest=None,kind='export_evidence', **{k:source.binding[k] for k in ('origin','service_id','issuer','delegation_id')}, endpoint=source.base+'exports', key_id=source.service.key_id,key_use=p.TYPE,request=copy.deepcopy(f.request),offer_seconds=0, access_seconds=0,use_seconds=0,payment_mode='none',asset_id=None,version_id=None,agreement_id=source.agreement_id) with source.store.connect() as db: keys = {r['key_id']:dict(p._key_descriptor(r),revoked=False) for r in db.execute('SELECT * FROM clients WHERE jwk IS NOT NULL')} pins = {crypto.digest(raw):raw for raw in (snapshot.authority_bytes,snapshot.policy_bytes)} for observed in source.history.observations(f.origin,limit=1000): historical = source.history.history(f.origin,observed['authority_revision'],observed['policy_revision']) for name in ('authority_bytes','policy_bytes'): raw = bytes(historical[name]); pins[crypto.digest(raw)] = raw trust = p.ExternalTrust(snapshot,pins,keys) selected = storage if selected_storage else p body = selected.build_bundle(source.store,agreement_id=source.agreement_id,snapshot=snapshot,history=source.history, export_operation=operation,private_key=source.service.private_key,key_id=source.service.key_id,now=f.now) archive = (storage.StorageEvidenceArchive if selected_storage else p.EvidenceArchive)(f.path/'retained'/'archive.sqlite3') report = archive.import_bundle(body,trust=trust,now=f.now) return archive, body, report, trust def test_real_cutover_preserves_exact_evidence_and_requires_fresh_assent(self): f = self.f f.policy['rules'][0]['obligations'].append(dict(type='retention',max_seconds=120)) f.request['supported_obligations'].append('retention') f.agree(wire=True); old_token = f.token(wire=True) old_receipt, old_offer = f.receipt_jws, f.offer_raw old_gateway = self.gateway(f,'old-gateway'); self.selected_gateway = old_gateway; self.resource(old_token) archive, body, report, trust = self.export_old() b = self.prepare(); self.publish(); self.agree_b() f.offset = 2; self.publish(withdraw=True) self.selected_gateway = self.gateway(b,'successor-gateway') self.resource(old_token,expected=403); self.resource(self.token_b()) self.assertNotEqual(b.agreement_id,f.agreement_id) self.assertNotEqual(b.receipt['service_id'],f.receipt['service_id']) self.assertEqual(self.call_b('agreements',b.assent),b.receipt_jws) self.http(f.host,urlsplit(f.base).path+'tokens',body=dict(f.binding,agreement_id=f.agreement_id), auth=f.clients['agent']['authorization'],expected=503) self.selected_gateway = old_gateway before = len(f.provider_server.requests); self.resource(old_token,expected=403) self.assertEqual(len(f.provider_server.requests),before) saved = archive.get(report['origin'],report['bundle_id']) self.assertEqual(saved['bundle_bytes'],body); self.assertEqual(saved['authority_status'],'historical_only') self.assertFalse(saved['reactivates_access']) _, receipt = f.call(f'agreements/{f.agreement_id}/receipt',method='GET',wire=True,status=200) self.assertEqual(receipt,old_receipt); self.assertEqual(f.offer_raw,old_offer) current = f.transport.fetch(f.origin) with self.assertRaises(ValueError): p.verify_bundle(body,trust=p.ExternalTrust(current,trust.pinned_documents,trust.client_keys),now=f.now) def test_old_collector_can_drain_selected_storage_without_future_access_authority(self): f = self.f; configure(f); agree(f) original_offer, original_receipt = f.offer_raw, f.receipt_jws write = action(f); start = session(write) for event in (write,start): f.call('events',signed(f,event),wire=True,status=201) b = self.prepare(); self.publish() # Only the delegation is reduced. The service/key identity is retained # unchanged; a key's uses cannot be rewritten under its existing kid. old_grant = f.authority['delegations'][0] old_grant.update(capabilities=['receive_events','export_evidence'],max_access_seconds=0,max_use_seconds=0,payment_mode='none') f.authority['revision'] += 1 f.call('tokens',dict(f.binding,agreement_id=f.agreement_id),status=503) f.offset = 2 end = copy.deepcopy(start); end.update(event_id=uid(),occurred_at=f.now()) end['operation'].update(state='completed',ended_at=f.now(), storage=dict(trigger='cessation',checkpoint_index=None,scheduled_at=f.now())) f.call('events',signed(f,end),wire=True,status=201) archive, body, report, _ = self.export_old(selected_storage=True) self.assertEqual(report['storage_metrics']['sessions_closed'],1) self.assertEqual(report['storage_metrics']['unresolved_sessions'],0) self.assertEqual(end['operation']['copy_context'],write['operation']['copy_context']) self.publish(withdraw=True) unseen = copy.deepcopy(end); unseen['event_id'] = uid() f.call('events',signed(f,unseen),status=503) self.assertEqual(archive.get(report['origin'],report['bundle_id'])['bundle_bytes'],body) with f.store.connect() as db: self.assertEqual(bytes(db.execute('SELECT body FROM offers').fetchone()[0]),original_offer) self.assertEqual(db.execute('SELECT receipt FROM agreements').fetchone()[0].encode(),original_receipt) def test_successor_staging_cannot_self_appoint_or_receive_old_credentials(self): f = self.f; f.agree(wire=True); b = self.prepare() request = dict(b.binding,principal_id=b.principal,request=f.request,access_seconds=120,use_seconds=240) self.call_b('offers',request,expected=503) self.publish() self.call_b('offers',request,auth=f.clients['agent']['authorization'],expected=401) before = len(self.successor_server.requests) with self.assertRaises(ValueError): f.transport.request(b.host,urlsplit(b.base).path+'offers',method='POST', body=crypto.json_bytes(request),credential=ScopedCredential(f.host,urlsplit(f.base).path,f.clients['agent']['authorization'])) self.assertEqual(len(self.successor_server.requests),before) self.call_b('agreements',f.accept_jws,expected=400) self.call_b('tokens',dict(b.binding,agreement_id=f.agreement_id),expected=404) with self.assertRaises(ValueError): FreeStore(f.store.path,{k:b.binding[k] for k in ('origin','service_id','issuer')}) original = self.gateway(f,'fixed-gateway') with self.assertRaises(ValueError): FreeGateway(original.path,binding=b.binding,base_url=b.base, transport=f.transport,history=b.history,credential=ScopedCredential(b.host,urlsplit(b.base).path,b.clients['gateway']['authorization']),clock=f.now) for _,path,headers,_ in f.origin_server.requests: if path in {'/odexa.json','/odexa-service.json'}: self.assertNotIn('Authorization',headers) def test_outage_and_rollback_after_cutover_never_fall_back_to_old_provider(self): f = self.f; f.agree(wire=True); old = f.token(wire=True); original = copy.deepcopy(f.authority) b = self.prepare(); self.publish(); self.agree_b(); f.offset = 2; self.publish(withdraw=True) self.selected_gateway = self.gateway(b,'successor-gateway'); token = self.token_b(); self.resource(token) final = copy.deepcopy(f.authority); before = len(f.provider_server.requests) self.successor_server.route = lambda *_: (503,{'Content-Type':'application/json'},b'{}') self.resource(token,expected=403) self.successor_server.route = b.service.dispatch f.origin_status = 503; self.resource(token,expected=403); self.resource(old,expected=403) f.origin_status = 200; f.authority = original; self.resource(token,expected=403) self.token_b(expected=503) self.assertEqual(len(f.provider_server.requests),before) f.authority = final; self.resource(token) self.selected_gateway = FreeGateway(self.selected_gateway.path,binding=b.binding,base_url=b.base, transport=f.transport,history=AuthorityHistory(f.path/'successor-gateway-history.sqlite3'), credential=ScopedCredential(b.host,urlsplit(b.base).path,b.clients['gateway']['authorization']),clock=f.now) f.authority = original; self.resource(token,expected=403) def test_live_key_rotation_retains_old_receipt_and_blocks_retired_signer(self): f = self.f; b = self.prepare(); self.publish(); self.agree_b() original = b.receipt_jws; retired_signer = b.service.private_key; retired_id = b.service.key_id next_key = ec.derive_private_key(76,ec.SECP256R1()); next_id = b.host+'/keys/two' b.descriptor['signing_keys'].append(dict(b.descriptor['signing_keys'][0],kid=next_id,public_jwk=crypto.public_jwk(next_key))) f.authority['revision'] += 1 # Publish the new key before it is used, then retire the predecessor. b.history.observe(f.transport.fetch(f.origin),now=f.now) b.service = FreeAgreementService(b.store,**dict(b.args,key_id=next_id,private_key=next_key)) self.successor_server.route = b.service.dispatch f.offset = 2; b.descriptor['signing_keys'][0].update(state='retired',retired_at=f.now()); f.authority['revision'] += 1 status = self.call_b(f'agreements/{b.agreement_id}/status',method='GET') crypto.verify_jws(status,crypto.public_jwk(next_key),'odexa-status+jws',next_id) self.assertEqual(self.call_b(f'agreements/{b.agreement_id}/receipt',method='GET'),original) self.assertEqual(self.call_b('agreements',b.assent),original) b.service.private_key = retired_signer; b.service.public_jwk = crypto.public_jwk(retired_signer); b.service.key_id = retired_id self.call_b(f'agreements/{b.agreement_id}/status',method='GET',expected=503) b.service.private_key = next_key; b.service.public_jwk = crypto.public_jwk(next_key); b.service.key_id = next_id self.assertIsNotNone(self.token_b()) # The currently authorised exporter is the NEW key. Retirement of the # old receipt key passes; known compromise of that historical key fails. archive, body, report, trust = self.export_old(who=b) f.offset = 3 b.descriptor['signing_keys'][0].update(state='revoked',revoked_at=f.now()) f.authority['revision'] += 1 current = f.transport.fetch(f.origin) with self.assertRaisesRegex(ValueError,'known compromised material'): p.verify_bundle(body,trust=p.ExternalTrust(current,trust.pinned_documents,trust.client_keys),now=f.now) self.assertEqual(archive.get(report['origin'],report['bundle_id'])['authority_status'],'historical_only') def test_paid_rights_stay_original_and_new_provider_requires_new_quote_and_mandate(self): self.f.tearDown() old = paid_fixture.PaidEvidenceTests(); old.setUp(); self.f = old.f; f = self.f f.origin_server.route = self.origin_route old.agree(); old.t.mandate(); old.check('confirmed'); old_token = f.token(wire=True) body, trust = old.export(); archive = paid.PaidEvidenceArchive(f.path/'retained'/'paid.sqlite3') report = archive.import_bundle(body,trust=trust,now=f.now) old_receipt = f.receipt_jws; old_quote = old.t.quote_raw; old_mandate = old.t.mandate_jws b = self.prepare(paid_fixture=old.t); self.publish(); self.agree_b(paid_mode=True) self.token_b(expected=403); self.assertEqual(b.verify_calls,0) route = f'agreements/{b.agreement_id}/payment-mandate' self.call_b(route,old_mandate,auth=old.t.payer_auth,expected=401) self.call_b(route,old_mandate,auth=b.payer_auth,expected=400) with b.store.connect() as db: qraw = bytes(db.execute('SELECT body FROM documents WHERE digest=?',(b.offer['payment']['quote_digest'],)).fetchone()[0]) q = pc.bind_quote(b.offer,qraw); self.assertNotEqual(qraw,old_quote) mandate = dict(old.t.mandate_doc,mandate_id=uid(),agreement_id=b.agreement_id,quote_id=q['quote_id'], quote_digest=crypto.digest(qraw),service_id=b.binding['service_id'],provider_id=b.verifier['issuer'], issued_at=f.now(),expires_at=later(f.now(),120)) signed = crypto.sign_jws(crypto.json_bytes(mandate),b.payer_key,b.payer_kid,pc.MANDATE_TYPE,allowed_types={pc.MANDATE_TYPE}).encode() self.call_b(route,signed,auth=b.payer_auth,expected=201) f.offset = 2; self.publish(withdraw=True) self.selected_gateway = self.gateway(b,'paid-successor-gateway') self.resource(old_token,expected=403); self.resource(self.token_b()) self.assertGreater(b.verify_calls,0) with f.store.connect() as db: previous = TransactionPayments(db)._read(f.agreement_id) self.assertEqual(previous.payment_state,'confirmed') self.assertEqual(previous.quote_bytes,old_quote) self.assertEqual(db.execute('SELECT receipt FROM agreements').fetchone()[0].encode(),old_receipt) saved = archive.get(report['origin'],report['bundle_id']) self.assertEqual(saved['bundle_bytes'],body); self.assertFalse(saved['reactivates_access']) self.assertEqual(saved['import_report']['payment']['payment_state'],'confirmed') self.assertFalse(saved['import_report']['payment']['bank_settlement_verified']) if __name__ == '__main__': unittest.main()