feat(tokens-repo):empty implementation of redis token repo

But it initializes and fails tests!
This commit is contained in:
Houkime 2022-12-14 14:29:19 +00:00
parent ca822cdf6f
commit f2fa47466b
2 changed files with 56 additions and 4 deletions

View file

@ -1,9 +1,18 @@
"""
Token repository using Redis as backend.
"""
from typing import Optional
from datetime import datetime
from selfprivacy_api.repositories.tokens.abstract_tokens_repository import (
AbstractTokensRepository,
)
from selfprivacy_api.utils.redis_pool import RedisPool
from selfprivacy_api.models.tokens.token import Token
from selfprivacy_api.models.tokens.recovery_key import RecoveryKey
from selfprivacy_api.models.tokens.new_device_key import NewDeviceKey
TOKENS_PREFIX = "token_repo:tokens:"
class RedisTokensRepository(AbstractTokensRepository):
@ -11,5 +20,48 @@ class RedisTokensRepository(AbstractTokensRepository):
Token repository using Redis as a backend
"""
def __init__(self) -> None:
def __init__(self):
self.connection = RedisPool().get_connection()
def token_key_for_device(device_name: str):
return TOKENS_PREFIX + str(hash(device_name))
def get_tokens(self) -> list[Token]:
"""Get the tokens"""
raise NotImplementedError
def delete_token(self, input_token: Token) -> None:
"""Delete the token"""
raise NotImplementedError
def get_recovery_key(self) -> Optional[RecoveryKey]:
"""Get the recovery key"""
raise NotImplementedError
def create_recovery_key(
self,
expiration: Optional[datetime],
uses_left: Optional[int],
) -> RecoveryKey:
"""Create the recovery key"""
raise NotImplementedError
def get_new_device_key(self) -> NewDeviceKey:
"""Creates and returns the new device key"""
raise NotImplementedError
def delete_new_device_key(self) -> None:
"""Delete the new device key"""
raise NotImplementedError
def _store_token(self, new_token: Token):
"""Store a token directly"""
raise NotImplementedError
def _decrement_recovery_token(self):
"""Decrement recovery key use count by one"""
raise NotImplementedError
def _get_stored_new_device_key(self) -> Optional[NewDeviceKey]:
"""Retrieves new device key that is already stored."""
raise NotImplementedError

View file

@ -166,12 +166,12 @@ def empty_redis_repo():
@pytest.fixture(params=["json", "redis"])
def empty_repo(request, empty_json_repo):
def empty_repo(request, empty_json_repo, empty_redis_repo):
if request.param == "json":
return empty_json_repo
if request.param == "redis":
# return empty_redis_repo
return empty_json_repo
return empty_redis_repo
# return empty_json_repo
else:
raise NotImplementedError