mirror of
https://git.collinwebdesigns.de/oscar.krause/fastapi-dls.git
synced 2025-08-26 22:47:00 +08:00
Compare commits
3 Commits
a7553716de
...
59beebdeff
Author | SHA1 | Date | |
---|---|---|---|
|
59beebdeff | ||
|
304ce0002c | ||
|
001d8a0ab1 |
13
app/main.py
13
app/main.py
@ -710,7 +710,7 @@ async def leasing_v1_lessor(request: Request):
|
||||
Lease.create_or_update(db, data)
|
||||
|
||||
response = {
|
||||
"client_challenge": None,
|
||||
"client_challenge": j.get('client_challenge'),
|
||||
"lease_result_list": lease_result_list,
|
||||
"result_code": None,
|
||||
"sync_timestamp": cur_time.strftime(DT_FORMAT),
|
||||
@ -719,7 +719,12 @@ async def leasing_v1_lessor(request: Request):
|
||||
|
||||
logger.debug(response)
|
||||
|
||||
return JSONr(response, headers={'X-NLS-Signature': '?'})
|
||||
si_certificate_filename = join(dirname(__file__), 'cert/my_demo_si_certificate.pem')
|
||||
my_si_certificate = Cert.from_file(si_certificate_filename)
|
||||
signature = my_si_certificate.signature().hex()
|
||||
signature = f'b\'{signature}\''
|
||||
|
||||
return JSONr(response, headers={'X-NLS-Signature': signature})
|
||||
|
||||
|
||||
# venv/lib/python3.9/site-packages/nls_services_lease/test/test_lease_multi_controller.py
|
||||
@ -746,7 +751,7 @@ async def leasing_v1_lessor_lease(request: Request):
|
||||
# venv/lib/python3.9/site-packages/nls_core_lease/lease_single.py
|
||||
@app.put('/leasing/v1/lease/{lease_ref}', description='renew a lease')
|
||||
async def leasing_v1_lease_renew(request: Request, lease_ref: str):
|
||||
token, cur_time = __get_token(request), datetime.now(UTC)
|
||||
j, token, cur_time = json_loads((await request.body()).decode('utf-8')), __get_token(request), datetime.now(UTC)
|
||||
|
||||
origin_ref = token.get('origin_ref')
|
||||
logger.info(f'> [ renew ]: {origin_ref}: renew {lease_ref}')
|
||||
@ -757,6 +762,7 @@ async def leasing_v1_lease_renew(request: Request, lease_ref: str):
|
||||
|
||||
expires = cur_time + LEASE_EXPIRE_DELTA
|
||||
response = {
|
||||
"client_challenge": j.get('client_challenge'),
|
||||
"lease_ref": lease_ref,
|
||||
"expires": expires.strftime(DT_FORMAT),
|
||||
"recommended_lease_renewal": LEASE_RENEWAL_PERIOD,
|
||||
@ -788,6 +794,7 @@ async def leasing_v1_lease_delete(request: Request, lease_ref: str):
|
||||
return JSONr(status_code=404, content={'status': 404, 'detail': 'lease not found'})
|
||||
|
||||
response = {
|
||||
"client_challenge": None,
|
||||
"lease_ref": lease_ref,
|
||||
"prompts": None,
|
||||
"sync_timestamp": cur_time.strftime(DT_FORMAT),
|
||||
|
@ -99,6 +99,9 @@ class Cert:
|
||||
def pem(self) -> bytes:
|
||||
return self.__cert.public_bytes(encoding=serialization.Encoding.PEM)
|
||||
|
||||
def signature(self) -> bytes:
|
||||
return self.__cert.signature
|
||||
|
||||
|
||||
def load_file(filename: str) -> bytes:
|
||||
log = logging.getLogger(f'{__name__}')
|
||||
|
10
test/main.py
10
test/main.py
@ -195,6 +195,7 @@ def test_auth_v1_token():
|
||||
|
||||
def test_leasing_v1_lessor():
|
||||
payload = {
|
||||
'client_challenge': 'my_unique_string',
|
||||
'fulfillment_context': {
|
||||
'fulfillment_class_ref_list': []
|
||||
},
|
||||
@ -209,6 +210,9 @@ def test_leasing_v1_lessor():
|
||||
response = client.post('/leasing/v1/lessor', json=payload, headers={'authorization': __bearer_token(ORIGIN_REF)})
|
||||
assert response.status_code == 200
|
||||
|
||||
client_challenge = response.json().get('client_challenge')
|
||||
assert client_challenge == payload.get('client_challenge')
|
||||
|
||||
lease_result_list = response.json().get('lease_result_list')
|
||||
assert len(lease_result_list) == 1
|
||||
assert len(lease_result_list[0]['lease']['ref']) == 36
|
||||
@ -232,9 +236,13 @@ def test_leasing_v1_lease_renew():
|
||||
|
||||
###
|
||||
|
||||
response = client.put(f'/leasing/v1/lease/{active_lease_ref}', headers={'authorization': __bearer_token(ORIGIN_REF)})
|
||||
payload = {'client_challenge': 'my_unique_string'}
|
||||
response = client.put(f'/leasing/v1/lease/{active_lease_ref}', json=payload, headers={'authorization': __bearer_token(ORIGIN_REF)})
|
||||
assert response.status_code == 200
|
||||
|
||||
client_challenge = response.json().get('client_challenge')
|
||||
assert client_challenge == payload.get('client_challenge')
|
||||
|
||||
lease_ref = response.json().get('lease_ref')
|
||||
assert len(lease_ref) == 36
|
||||
assert lease_ref == active_lease_ref
|
||||
|
Loading…
x
Reference in New Issue
Block a user