From f2fa47466bf3d8aab6f857cd4291fe2034750838 Mon Sep 17 00:00:00 2001 From: Houkime <> Date: Wed, 14 Dec 2022 14:29:19 +0000 Subject: [PATCH] feat(tokens-repo):empty implementation of redis token repo But it initializes and fails tests! --- .../tokens/redis_tokens_repository.py | 54 ++++++++++++++++++- .../test_repository/test_tokens_repository.py | 6 +-- 2 files changed, 56 insertions(+), 4 deletions(-) diff --git a/selfprivacy_api/repositories/tokens/redis_tokens_repository.py b/selfprivacy_api/repositories/tokens/redis_tokens_repository.py index 0186c11..13d49c9 100644 --- a/selfprivacy_api/repositories/tokens/redis_tokens_repository.py +++ b/selfprivacy_api/repositories/tokens/redis_tokens_repository.py @@ -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 diff --git a/tests/test_graphql/test_repository/test_tokens_repository.py b/tests/test_graphql/test_repository/test_tokens_repository.py index bdad6d8..dff1799 100644 --- a/tests/test_graphql/test_repository/test_tokens_repository.py +++ b/tests/test_graphql/test_repository/test_tokens_repository.py @@ -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