fix(tokens): normalize exceptions, fix KeyErrors
parent
b31c06a0f7
commit
042f2b2310
|
@ -1,18 +1,14 @@
|
||||||
class TokenNotFoundError(Exception):
|
class TokenNotFound(Exception):
|
||||||
"""Token not found!"""
|
"""Token not found!"""
|
||||||
|
|
||||||
|
|
||||||
class RecoveryKeyNotFoundError(Exception):
|
class RecoveryKeyNotFound(Exception):
|
||||||
"""Recovery key not found!"""
|
"""Recovery key not found!"""
|
||||||
|
|
||||||
|
|
||||||
class MnemonicError(Exception):
|
class InvalidMnemonic(Exception):
|
||||||
"""Phrase is not mnemonic!"""
|
"""Phrase is not mnemonic!"""
|
||||||
|
|
||||||
|
|
||||||
class RecoveryKeyIsNotValidError(Exception):
|
class NewDeviceKeyNotFound(Exception):
|
||||||
"""Recovery key is not valid!"""
|
"""New device key not found!"""
|
||||||
|
|
||||||
|
|
||||||
class RecoveryTokenError(Exception):
|
|
||||||
"""Error ???"""
|
|
||||||
|
|
|
@ -3,22 +3,21 @@ temporary legacy
|
||||||
"""
|
"""
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
from mnemonic import Mnemonic
|
||||||
|
|
||||||
from selfprivacy_api.utils import UserDataFiles, WriteUserData, ReadUserData
|
from selfprivacy_api.utils import UserDataFiles, WriteUserData, ReadUserData
|
||||||
from selfprivacy_api.models.tokens.token import Token
|
from selfprivacy_api.models.tokens.token import Token
|
||||||
from selfprivacy_api.models.tokens.recovery_key import RecoveryKey
|
from selfprivacy_api.models.tokens.recovery_key import RecoveryKey
|
||||||
from selfprivacy_api.models.tokens.new_device_key import NewDeviceKey
|
from selfprivacy_api.models.tokens.new_device_key import NewDeviceKey
|
||||||
from selfprivacy_api.repositories.tokens.exceptions import (
|
from selfprivacy_api.repositories.tokens.exceptions import (
|
||||||
TokenNotFoundError,
|
TokenNotFound,
|
||||||
RecoveryKeyNotFoundError,
|
RecoveryKeyNotFound,
|
||||||
MnemonicError,
|
InvalidMnemonic,
|
||||||
RecoveryKeyIsNotValidError,
|
NewDeviceKeyNotFound,
|
||||||
RecoveryTokenError,
|
|
||||||
)
|
)
|
||||||
from selfprivacy_api.repositories.tokens.abstract_tokens_repository import (
|
from selfprivacy_api.repositories.tokens.abstract_tokens_repository import (
|
||||||
AbstractTokensRepository,
|
AbstractTokensRepository,
|
||||||
)
|
)
|
||||||
from mnemonic import Mnemonic
|
|
||||||
|
|
||||||
DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"
|
DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S.%f"
|
||||||
|
|
||||||
|
@ -36,7 +35,7 @@ class JsonTokensRepository(AbstractTokensRepository):
|
||||||
created_at=userdata_token["date"],
|
created_at=userdata_token["date"],
|
||||||
)
|
)
|
||||||
|
|
||||||
raise TokenNotFoundError("Token not found!")
|
raise TokenNotFound("Token not found!")
|
||||||
|
|
||||||
def get_token_by_name(self, token_name: str) -> Optional[Token]:
|
def get_token_by_name(self, token_name: str) -> Optional[Token]:
|
||||||
"""Get the token by name"""
|
"""Get the token by name"""
|
||||||
|
@ -50,7 +49,7 @@ class JsonTokensRepository(AbstractTokensRepository):
|
||||||
created_at=userdata_token["date"],
|
created_at=userdata_token["date"],
|
||||||
)
|
)
|
||||||
|
|
||||||
raise TokenNotFoundError("Token not found!")
|
raise TokenNotFound("Token not found!")
|
||||||
|
|
||||||
def get_tokens(self) -> list[Token]:
|
def get_tokens(self) -> list[Token]:
|
||||||
"""Get the tokens"""
|
"""Get the tokens"""
|
||||||
|
@ -90,7 +89,7 @@ class JsonTokensRepository(AbstractTokensRepository):
|
||||||
tokens_file["tokens"].remove(userdata_token)
|
tokens_file["tokens"].remove(userdata_token)
|
||||||
return
|
return
|
||||||
|
|
||||||
raise TokenNotFoundError("Token not found!")
|
raise TokenNotFound("Token not found!")
|
||||||
|
|
||||||
def refresh_token(self, input_token: Token) -> Token:
|
def refresh_token(self, input_token: Token) -> Token:
|
||||||
"""Change the token field of the existing token"""
|
"""Change the token field of the existing token"""
|
||||||
|
@ -107,13 +106,16 @@ class JsonTokensRepository(AbstractTokensRepository):
|
||||||
|
|
||||||
return new_token
|
return new_token
|
||||||
|
|
||||||
raise TokenNotFoundError("Token not found!")
|
raise TokenNotFound("Token not found!")
|
||||||
|
|
||||||
def get_recovery_key(self) -> Optional[RecoveryKey]:
|
def get_recovery_key(self) -> Optional[RecoveryKey]:
|
||||||
"""Get the recovery key"""
|
"""Get the recovery key"""
|
||||||
with ReadUserData(UserDataFiles.TOKENS) as tokens_file:
|
with ReadUserData(UserDataFiles.TOKENS) as tokens_file:
|
||||||
|
|
||||||
if tokens_file["recovery_token"] is None:
|
if (
|
||||||
|
"recovery_token" not in tokens_file
|
||||||
|
or tokens_file["recovery_token"] is None
|
||||||
|
):
|
||||||
return
|
return
|
||||||
|
|
||||||
recovery_key = RecoveryKey(
|
recovery_key = RecoveryKey(
|
||||||
|
@ -151,19 +153,19 @@ class JsonTokensRepository(AbstractTokensRepository):
|
||||||
recovery_key = self.get_recovery_key()
|
recovery_key = self.get_recovery_key()
|
||||||
|
|
||||||
if recovery_key is None:
|
if recovery_key is None:
|
||||||
raise RecoveryKeyNotFoundError("Recovery key is None!")
|
raise RecoveryKeyNotFound("Recovery key not found")
|
||||||
|
|
||||||
if not recovery_key.is_valid():
|
if not recovery_key.is_valid():
|
||||||
raise RecoveryKeyIsNotValidError("Recovery key is not valid!")
|
raise RecoveryKeyNotFound("Recovery key not found")
|
||||||
|
|
||||||
recovery_token = bytes.fromhex(recovery_key.key)
|
recovery_token = bytes.fromhex(recovery_key.key)
|
||||||
|
|
||||||
if not Mnemonic(language="english").check(mnemonic_phrase):
|
if not Mnemonic(language="english").check(mnemonic_phrase):
|
||||||
raise MnemonicError("Phrase is not mnemonic!")
|
raise InvalidMnemonic("Phrase is not mnemonic!")
|
||||||
|
|
||||||
phrase_bytes = Mnemonic(language="english").to_entropy(mnemonic_phrase)
|
phrase_bytes = Mnemonic(language="english").to_entropy(mnemonic_phrase)
|
||||||
if phrase_bytes != recovery_token:
|
if phrase_bytes != recovery_token:
|
||||||
raise RecoveryTokenError("Phrase is not recovery token")
|
raise RecoveryKeyNotFound("Recovery key not found")
|
||||||
|
|
||||||
new_token = Token.generate(device_name=device_name)
|
new_token = Token.generate(device_name=device_name)
|
||||||
|
|
||||||
|
@ -204,33 +206,29 @@ class JsonTokensRepository(AbstractTokensRepository):
|
||||||
del tokens_file["new_device"]
|
del tokens_file["new_device"]
|
||||||
return
|
return
|
||||||
|
|
||||||
raise TokenNotFoundError("Key not found!")
|
|
||||||
|
|
||||||
def use_mnemonic_new_device_key(
|
def use_mnemonic_new_device_key(
|
||||||
self, mnemonic_phrase: str, device_name: str
|
self, mnemonic_phrase: str, device_name: str
|
||||||
) -> Token:
|
) -> Token:
|
||||||
"""Use the mnemonic new device key"""
|
"""Use the mnemonic new device key"""
|
||||||
|
|
||||||
with WriteUserData(UserDataFiles.TOKENS) as tokens_file:
|
with ReadUserData(UserDataFiles.TOKENS) as tokens_file:
|
||||||
|
if "new_device" not in tokens_file or tokens_file["new_device"] is None:
|
||||||
|
raise NewDeviceKeyNotFound("New device key not found")
|
||||||
|
|
||||||
new_device_key = NewDeviceKey(
|
new_device_key = NewDeviceKey(
|
||||||
key=tokens_file["new_device"]["token"],
|
key=tokens_file["new_device"]["token"],
|
||||||
created_at=tokens_file["new_device"]["date"],
|
created_at=tokens_file["new_device"]["date"],
|
||||||
expires_at=tokens_file["new_device"]["expiration"],
|
expires_at=tokens_file["new_device"]["expiration"],
|
||||||
)
|
)
|
||||||
|
|
||||||
if new_device_key is None:
|
|
||||||
raise TokenNotFoundError("New device key not found!")
|
|
||||||
|
|
||||||
token = bytes.fromhex(new_device_key.key)
|
token = bytes.fromhex(new_device_key.key)
|
||||||
|
|
||||||
if not Mnemonic(language="english").check(mnemonic_phrase):
|
if not Mnemonic(language="english").check(mnemonic_phrase):
|
||||||
raise MnemonicError("Phrase is not mnemonic!")
|
raise InvalidMnemonic("Phrase is not mnemonic!")
|
||||||
|
|
||||||
phrase_bytes = Mnemonic(language="english").to_entropy(mnemonic_phrase)
|
phrase_bytes = Mnemonic(language="english").to_entropy(mnemonic_phrase)
|
||||||
if bytes(phrase_bytes) != bytes(
|
if bytes(phrase_bytes) != bytes(token):
|
||||||
token
|
raise NewDeviceKeyNotFound("Phrase is not token!")
|
||||||
): # idk why, но оно не робит, хотя оригинальную логику я сохранил
|
|
||||||
raise TokenNotFoundError("Phrase is not token!")
|
|
||||||
|
|
||||||
new_token = Token.generate(device_name=device_name)
|
new_token = Token.generate(device_name=device_name)
|
||||||
with WriteUserData(UserDataFiles.TOKENS) as tokens:
|
with WriteUserData(UserDataFiles.TOKENS) as tokens:
|
||||||
|
|
|
@ -10,11 +10,10 @@ from selfprivacy_api.models.tokens.new_device_key import NewDeviceKey
|
||||||
from selfprivacy_api.models.tokens.recovery_key import RecoveryKey
|
from selfprivacy_api.models.tokens.recovery_key import RecoveryKey
|
||||||
from selfprivacy_api.models.tokens.token import Token
|
from selfprivacy_api.models.tokens.token import Token
|
||||||
from selfprivacy_api.repositories.tokens.exceptions import (
|
from selfprivacy_api.repositories.tokens.exceptions import (
|
||||||
MnemonicError,
|
InvalidMnemonic,
|
||||||
RecoveryKeyIsNotValidError,
|
RecoveryKeyNotFound,
|
||||||
RecoveryKeyNotFoundError,
|
TokenNotFound,
|
||||||
RecoveryTokenError,
|
NewDeviceKeyNotFound,
|
||||||
TokenNotFoundError,
|
|
||||||
)
|
)
|
||||||
from selfprivacy_api.repositories.tokens.json_tokens_repository import (
|
from selfprivacy_api.repositories.tokens.json_tokens_repository import (
|
||||||
JsonTokensRepository,
|
JsonTokensRepository,
|
||||||
|
@ -22,6 +21,58 @@ from selfprivacy_api.repositories.tokens.json_tokens_repository import (
|
||||||
from tests.common import read_json
|
from tests.common import read_json
|
||||||
|
|
||||||
|
|
||||||
|
ORIGINAL_TOKEN_CONTENT = [
|
||||||
|
{
|
||||||
|
"token": "KG9ni-B-CMPk327Zv1qC7YBQaUGaBUcgdkvMvQ2atFI",
|
||||||
|
"name": "primary_token",
|
||||||
|
"date": "2022-07-15 17:41:31.675698",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"token": "3JKgLOtFu6ZHgE4OU-R-VdW47IKpg-YQL0c6n7bol68",
|
||||||
|
"name": "second_token",
|
||||||
|
"date": "2022-07-15 17:41:31.675698Z",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"token": "LYiwFDekvALKTQSjk7vtMQuNP_6wqKuV-9AyMKytI_8",
|
||||||
|
"name": "third_token",
|
||||||
|
"date": "2022-07-15T17:41:31.675698Z",
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"token": "dD3CFPcEZvapscgzWb7JZTLog7OMkP7NzJeu2fAazXM",
|
||||||
|
"name": "forth_token",
|
||||||
|
"date": "2022-07-15T17:41:31.675698",
|
||||||
|
},
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def tokens(mocker, datadir):
|
||||||
|
mocker.patch("selfprivacy_api.utils.TOKENS_FILE", new=datadir / "tokens.json")
|
||||||
|
assert read_json(datadir / "tokens.json")["tokens"] == ORIGINAL_TOKEN_CONTENT
|
||||||
|
return datadir
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def empty_keys(mocker, datadir):
|
||||||
|
mocker.patch("selfprivacy_api.utils.TOKENS_FILE", new=datadir / "empty_keys.json")
|
||||||
|
assert read_json(datadir / "empty_keys.json")["tokens"] == [
|
||||||
|
{
|
||||||
|
"token": "KG9ni-B-CMPk327Zv1qC7YBQaUGaBUcgdkvMvQ2atFI",
|
||||||
|
"name": "primary_token",
|
||||||
|
"date": "2022-07-15 17:41:31.675698",
|
||||||
|
}
|
||||||
|
]
|
||||||
|
return datadir
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def null_keys(mocker, datadir):
|
||||||
|
mocker.patch("selfprivacy_api.utils.TOKENS_FILE", new=datadir / "null_keys.json")
|
||||||
|
assert read_json(datadir / "null_keys.json")["recovery_token"] is None
|
||||||
|
assert read_json(datadir / "null_keys.json")["new_device"] is None
|
||||||
|
return datadir
|
||||||
|
|
||||||
|
|
||||||
class RecoveryKeyMockReturnNotValid:
|
class RecoveryKeyMockReturnNotValid:
|
||||||
def is_valid() -> bool:
|
def is_valid() -> bool:
|
||||||
return False
|
return False
|
||||||
|
@ -41,16 +92,6 @@ def mock_new_device_key_generate(mocker):
|
||||||
return mock
|
return mock
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def mock_get_recovery_key_return_none(mocker):
|
|
||||||
mock = mocker.patch(
|
|
||||||
"selfprivacy_api.repositories.tokens.json_tokens_repository.JsonTokensRepository.get_recovery_key",
|
|
||||||
autospec=True,
|
|
||||||
return_value=None,
|
|
||||||
)
|
|
||||||
return mock
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def mock_generate_token(mocker):
|
def mock_generate_token(mocker):
|
||||||
mock = mocker.patch(
|
mock = mocker.patch(
|
||||||
|
@ -104,32 +145,9 @@ def mock_recovery_key_generate(mocker):
|
||||||
return mock
|
return mock
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
###############
|
||||||
def tokens(mocker, datadir):
|
# Test tokens #
|
||||||
mocker.patch("selfprivacy_api.utils.TOKENS_FILE", new=datadir / "tokens.json")
|
###############
|
||||||
assert read_json(datadir / "tokens.json")["tokens"] == [
|
|
||||||
{
|
|
||||||
"token": "KG9ni-B-CMPk327Zv1qC7YBQaUGaBUcgdkvMvQ2atFI",
|
|
||||||
"name": "primary_token",
|
|
||||||
"date": "2022-07-15 17:41:31.675698",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"token": "3JKgLOtFu6ZHgE4OU-R-VdW47IKpg-YQL0c6n7bol68",
|
|
||||||
"name": "second_token",
|
|
||||||
"date": "2022-07-15 17:41:31.675698Z",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"token": "LYiwFDekvALKTQSjk7vtMQuNP_6wqKuV-9AyMKytI_8",
|
|
||||||
"name": "third_token",
|
|
||||||
"date": "2022-07-15T17:41:31.675698Z",
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"token": "dD3CFPcEZvapscgzWb7JZTLog7OMkP7NzJeu2fAazXM",
|
|
||||||
"name": "forth_token",
|
|
||||||
"date": "2022-07-15T17:41:31.675698",
|
|
||||||
},
|
|
||||||
]
|
|
||||||
return datadir
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_token_by_token_string(tokens):
|
def test_get_token_by_token_string(tokens):
|
||||||
|
@ -147,7 +165,7 @@ def test_get_token_by_token_string(tokens):
|
||||||
def test_get_token_by_non_existent_token_string(tokens):
|
def test_get_token_by_non_existent_token_string(tokens):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
with pytest.raises(TokenNotFoundError):
|
with pytest.raises(TokenNotFound):
|
||||||
assert repo.get_token_by_token_string(token_string="iamBadtoken") is None
|
assert repo.get_token_by_token_string(token_string="iamBadtoken") is None
|
||||||
|
|
||||||
|
|
||||||
|
@ -165,14 +183,13 @@ def test_get_token_by_name(tokens):
|
||||||
def test_get_token_by_non_existent_name(tokens):
|
def test_get_token_by_non_existent_name(tokens):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
with pytest.raises(TokenNotFoundError):
|
with pytest.raises(TokenNotFound):
|
||||||
assert repo.get_token_by_name(token_name="badname") is None
|
assert repo.get_token_by_name(token_name="badname") is None
|
||||||
|
|
||||||
|
|
||||||
def test_get_tokens(tokens):
|
def test_get_tokens(tokens):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
assert repo.get_tokens() is not None
|
|
||||||
assert repo.get_tokens() == [
|
assert repo.get_tokens() == [
|
||||||
Token(
|
Token(
|
||||||
token="KG9ni-B-CMPk327Zv1qC7YBQaUGaBUcgdkvMvQ2atFI",
|
token="KG9ni-B-CMPk327Zv1qC7YBQaUGaBUcgdkvMvQ2atFI",
|
||||||
|
@ -197,6 +214,18 @@ def test_get_tokens(tokens):
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_tokens_when_one(empty_keys):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
assert repo.get_tokens() == [
|
||||||
|
Token(
|
||||||
|
token="KG9ni-B-CMPk327Zv1qC7YBQaUGaBUcgdkvMvQ2atFI",
|
||||||
|
device_name="primary_token",
|
||||||
|
created_at=datetime(2022, 7, 15, 17, 41, 31, 675698),
|
||||||
|
)
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
def test_create_token(tokens, mock_token_generate):
|
def test_create_token(tokens, mock_token_generate):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
@ -207,7 +236,7 @@ def test_create_token(tokens, mock_token_generate):
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_delete_token(tokens, datadir):
|
def test_delete_token(tokens):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
input_token = Token(
|
input_token = Token(
|
||||||
token="KG9ni-B-CMPk327Zv1qC7YBQaUGaBUcgdkvMvQ2atFI",
|
token="KG9ni-B-CMPk327Zv1qC7YBQaUGaBUcgdkvMvQ2atFI",
|
||||||
|
@ -215,8 +244,8 @@ def test_delete_token(tokens, datadir):
|
||||||
created_at=datetime(2022, 7, 15, 17, 41, 31, 675698),
|
created_at=datetime(2022, 7, 15, 17, 41, 31, 675698),
|
||||||
)
|
)
|
||||||
|
|
||||||
assert repo.delete_token(input_token) is None
|
repo.delete_token(input_token)
|
||||||
assert read_json(datadir / "tokens.json")["tokens"] == [
|
assert read_json(tokens / "tokens.json")["tokens"] == [
|
||||||
{
|
{
|
||||||
"token": "3JKgLOtFu6ZHgE4OU-R-VdW47IKpg-YQL0c6n7bol68",
|
"token": "3JKgLOtFu6ZHgE4OU-R-VdW47IKpg-YQL0c6n7bol68",
|
||||||
"name": "second_token",
|
"name": "second_token",
|
||||||
|
@ -235,16 +264,18 @@ def test_delete_token(tokens, datadir):
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
def test_delete_not_found_token(tokens, datadir):
|
def test_delete_not_found_token(tokens):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
input_token = Token(
|
input_token = Token(
|
||||||
token="imbadtoken",
|
token="imbadtoken",
|
||||||
device_name="primary_token",
|
device_name="primary_token",
|
||||||
created_at=datetime(2022, 7, 15, 17, 41, 31, 675698),
|
created_at=datetime(2022, 7, 15, 17, 41, 31, 675698),
|
||||||
)
|
)
|
||||||
with pytest.raises(TokenNotFoundError):
|
with pytest.raises(TokenNotFound):
|
||||||
assert repo.delete_token(input_token) is None
|
assert repo.delete_token(input_token) is None
|
||||||
|
|
||||||
|
assert read_json(tokens / "tokens.json")["tokens"] == ORIGINAL_TOKEN_CONTENT
|
||||||
|
|
||||||
|
|
||||||
def test_refresh_token(tokens, mock_token_generate):
|
def test_refresh_token(tokens, mock_token_generate):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
|
@ -269,14 +300,18 @@ def test_refresh_not_found_token(tokens, mock_token_generate):
|
||||||
created_at=datetime(2022, 7, 15, 17, 41, 31, 675698),
|
created_at=datetime(2022, 7, 15, 17, 41, 31, 675698),
|
||||||
)
|
)
|
||||||
|
|
||||||
with pytest.raises(TokenNotFoundError):
|
with pytest.raises(TokenNotFound):
|
||||||
assert repo.refresh_token(input_token) is None
|
assert repo.refresh_token(input_token) is None
|
||||||
|
|
||||||
|
|
||||||
|
################
|
||||||
|
# Recovery key #
|
||||||
|
################
|
||||||
|
|
||||||
|
|
||||||
def test_get_recovery_key(tokens):
|
def test_get_recovery_key(tokens):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
assert repo.get_recovery_key() is not None
|
|
||||||
assert repo.get_recovery_key() == RecoveryKey(
|
assert repo.get_recovery_key() == RecoveryKey(
|
||||||
key="ed653e4b8b042b841d285fa7a682fa09e925ddb2d8906f54",
|
key="ed653e4b8b042b841d285fa7a682fa09e925ddb2d8906f54",
|
||||||
created_at=datetime(2022, 11, 11, 11, 48, 54, 228038),
|
created_at=datetime(2022, 11, 11, 11, 48, 54, 228038),
|
||||||
|
@ -285,11 +320,17 @@ def test_get_recovery_key(tokens):
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_create_recovery_key(tokens, mock_recovery_key_generate, datadir):
|
def test_get_recovery_key_when_empty(empty_keys):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
assert repo.get_recovery_key() is None
|
||||||
|
|
||||||
|
|
||||||
|
def test_create_recovery_key(tokens, mock_recovery_key_generate):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
assert repo.create_recovery_key(uses_left=1, expiration=None) is not None
|
assert repo.create_recovery_key(uses_left=1, expiration=None) is not None
|
||||||
assert read_json(datadir / "tokens.json")["recovery_token"] == {
|
assert read_json(tokens / "tokens.json")["recovery_token"] == {
|
||||||
"token": "889bf49c1d3199d71a2e704718772bd53a422020334db051",
|
"token": "889bf49c1d3199d71a2e704718772bd53a422020334db051",
|
||||||
"date": "2022-07-15T17:41:31.675698",
|
"date": "2022-07-15T17:41:31.675698",
|
||||||
"expiration": None,
|
"expiration": None,
|
||||||
|
@ -297,122 +338,12 @@ def test_create_recovery_key(tokens, mock_recovery_key_generate, datadir):
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
def test_get_new_device_key(tokens, mock_new_device_key_generate, datadir):
|
def test_use_mnemonic_recovery_key_when_empty(
|
||||||
repo = JsonTokensRepository()
|
empty_keys, mock_recovery_key_generate, mock_token_generate
|
||||||
|
|
||||||
assert repo.get_new_device_key() is not None
|
|
||||||
assert read_json(datadir / "tokens.json")["new_device"] == {
|
|
||||||
"date": "2022-07-15T17:41:31.675698",
|
|
||||||
"expiration": "2022-07-15T17:41:31.675698",
|
|
||||||
"token": "43478d05b35e4781598acd76e33832bb",
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def test_delete_new_device_key(tokens, datadir):
|
|
||||||
repo = JsonTokensRepository()
|
|
||||||
|
|
||||||
assert repo.delete_new_device_key() is None
|
|
||||||
assert "new_device" not in read_json(datadir / "tokens.json")
|
|
||||||
|
|
||||||
|
|
||||||
####################################################
|
|
||||||
|
|
||||||
|
|
||||||
def test_use_bad_mnemonic_phrase_new_device_key(
|
|
||||||
tokens, mock_new_device_key_generate, datadir, mock_token_generate
|
|
||||||
):
|
):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
with pytest.raises(MnemonicError):
|
with pytest.raises(RecoveryKeyNotFound):
|
||||||
assert (
|
|
||||||
repo.use_mnemonic_new_device_key(
|
|
||||||
device_name="imnew",
|
|
||||||
mnemonic_phrase="oh-no",
|
|
||||||
)
|
|
||||||
is None
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_use_not_exists_mnemonic_new_device_key(
|
|
||||||
tokens, mock_new_device_key_generate, datadir, mock_token_generate
|
|
||||||
):
|
|
||||||
repo = JsonTokensRepository()
|
|
||||||
|
|
||||||
with pytest.raises(TokenNotFoundError):
|
|
||||||
assert (
|
|
||||||
repo.use_mnemonic_new_device_key(
|
|
||||||
device_name="imnew",
|
|
||||||
mnemonic_phrase="uniform clarify napkin bid dress search input armor police cross salon because myself uphold slice bamboo hungry park",
|
|
||||||
)
|
|
||||||
is None
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_use_mnemonic_new_device_key(
|
|
||||||
tokens, mock_new_device_key_generate, datadir, mock_token_generate
|
|
||||||
):
|
|
||||||
repo = JsonTokensRepository()
|
|
||||||
|
|
||||||
assert (
|
|
||||||
repo.use_mnemonic_new_device_key(
|
|
||||||
device_name="imnew",
|
|
||||||
mnemonic_phrase="captain ribbon toddler settle symbol minute step broccoli bless universe divide bulb",
|
|
||||||
)
|
|
||||||
is not None
|
|
||||||
)
|
|
||||||
# assert read_json(datadir / "tokens.json")["new_device"] == []
|
|
||||||
|
|
||||||
|
|
||||||
def test_use_none_mnemonic_recovery_key(
|
|
||||||
datadir, tokens, mock_get_recovery_key_return_none
|
|
||||||
):
|
|
||||||
repo = JsonTokensRepository()
|
|
||||||
|
|
||||||
with pytest.raises(RecoveryKeyNotFoundError):
|
|
||||||
assert (
|
|
||||||
repo.use_mnemonic_recovery_key(
|
|
||||||
mnemonic_phrase="i love you",
|
|
||||||
device_name="primary_token",
|
|
||||||
)
|
|
||||||
is None
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_use_mnemonic_not_valid_recovery_key(
|
|
||||||
datadir, tokens, mock_get_recovery_key_return_not_valid
|
|
||||||
):
|
|
||||||
repo = JsonTokensRepository()
|
|
||||||
|
|
||||||
with pytest.raises(RecoveryKeyIsNotValidError):
|
|
||||||
assert (
|
|
||||||
repo.use_mnemonic_recovery_key(
|
|
||||||
mnemonic_phrase="sorry, it was joke",
|
|
||||||
device_name="primary_token",
|
|
||||||
)
|
|
||||||
is None
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_use_not_mnemonic_recovery_key(
|
|
||||||
datadir,
|
|
||||||
tokens,
|
|
||||||
):
|
|
||||||
repo = JsonTokensRepository()
|
|
||||||
|
|
||||||
with pytest.raises(MnemonicError):
|
|
||||||
assert (
|
|
||||||
repo.use_mnemonic_recovery_key(
|
|
||||||
mnemonic_phrase="please come back",
|
|
||||||
device_name="primary_token",
|
|
||||||
)
|
|
||||||
is None
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def test_use_not_found_mnemonic_recovery_key(datadir, tokens):
|
|
||||||
repo = JsonTokensRepository()
|
|
||||||
|
|
||||||
with pytest.raises(RecoveryTokenError):
|
|
||||||
assert (
|
assert (
|
||||||
repo.use_mnemonic_recovery_key(
|
repo.use_mnemonic_recovery_key(
|
||||||
mnemonic_phrase="captain ribbon toddler settle symbol minute step broccoli bless universe divide bulb",
|
mnemonic_phrase="captain ribbon toddler settle symbol minute step broccoli bless universe divide bulb",
|
||||||
|
@ -422,7 +353,87 @@ def test_use_not_found_mnemonic_recovery_key(datadir, tokens):
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_use_mnemonic_recovery_key(datadir, tokens, mock_generate_token):
|
def test_use_mnemonic_not_valid_recovery_key(
|
||||||
|
tokens, mock_get_recovery_key_return_not_valid
|
||||||
|
):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
with pytest.raises(RecoveryKeyNotFound):
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_recovery_key(
|
||||||
|
mnemonic_phrase="captain ribbon toddler settle symbol minute step broccoli bless universe divide bulb",
|
||||||
|
device_name="primary_token",
|
||||||
|
)
|
||||||
|
is None
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_mnemonic_not_mnemonic_recovery_key(tokens):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
with pytest.raises(InvalidMnemonic):
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_recovery_key(
|
||||||
|
mnemonic_phrase="sorry, it was joke",
|
||||||
|
device_name="primary_token",
|
||||||
|
)
|
||||||
|
is None
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_not_mnemonic_recovery_key(tokens):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
with pytest.raises(InvalidMnemonic):
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_recovery_key(
|
||||||
|
mnemonic_phrase="please come back",
|
||||||
|
device_name="primary_token",
|
||||||
|
)
|
||||||
|
is None
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_not_found_mnemonic_recovery_key(tokens):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
with pytest.raises(RecoveryKeyNotFound):
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_recovery_key(
|
||||||
|
mnemonic_phrase="captain ribbon toddler settle symbol minute step broccoli bless universe divide bulb",
|
||||||
|
device_name="primary_token",
|
||||||
|
)
|
||||||
|
is None
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_menemonic_recovery_key_when_empty(empty_keys):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
with pytest.raises(RecoveryKeyNotFound):
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_recovery_key(
|
||||||
|
mnemonic_phrase="captain ribbon toddler settle symbol minute step broccoli bless universe divide bulb",
|
||||||
|
device_name="primary_token",
|
||||||
|
)
|
||||||
|
is None
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_menemonic_recovery_key_when_null(null_keys):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
with pytest.raises(RecoveryKeyNotFound):
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_recovery_key(
|
||||||
|
mnemonic_phrase="captain ribbon toddler settle symbol minute step broccoli bless universe divide bulb",
|
||||||
|
device_name="primary_token",
|
||||||
|
)
|
||||||
|
is None
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_mnemonic_recovery_key(tokens, mock_generate_token):
|
||||||
repo = JsonTokensRepository()
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
assert repo.use_mnemonic_recovery_key(
|
assert repo.use_mnemonic_recovery_key(
|
||||||
|
@ -434,7 +445,7 @@ def test_use_mnemonic_recovery_key(datadir, tokens, mock_generate_token):
|
||||||
created_at=datetime(2022, 11, 14, 6, 6, 32, 777123),
|
created_at=datetime(2022, 11, 14, 6, 6, 32, 777123),
|
||||||
)
|
)
|
||||||
|
|
||||||
assert read_json(datadir / "tokens.json")["tokens"] == [
|
assert read_json(tokens / "tokens.json")["tokens"] == [
|
||||||
{
|
{
|
||||||
"date": "2022-07-15 17:41:31.675698",
|
"date": "2022-07-15 17:41:31.675698",
|
||||||
"name": "primary_token",
|
"name": "primary_token",
|
||||||
|
@ -462,9 +473,110 @@ def test_use_mnemonic_recovery_key(datadir, tokens, mock_generate_token):
|
||||||
},
|
},
|
||||||
]
|
]
|
||||||
|
|
||||||
assert read_json(datadir / "tokens.json")["recovery_token"] == {
|
assert read_json(tokens / "tokens.json")["recovery_token"] == {
|
||||||
"date": "2022-11-11T11:48:54.228038",
|
"date": "2022-11-11T11:48:54.228038",
|
||||||
"expiration": None,
|
"expiration": None,
|
||||||
"token": "ed653e4b8b042b841d285fa7a682fa09e925ddb2d8906f54",
|
"token": "ed653e4b8b042b841d285fa7a682fa09e925ddb2d8906f54",
|
||||||
"uses_left": 1,
|
"uses_left": 1,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
##################
|
||||||
|
# New device key #
|
||||||
|
##################
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_new_device_key(tokens, mock_new_device_key_generate):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
assert repo.get_new_device_key() is not None
|
||||||
|
assert read_json(tokens / "tokens.json")["new_device"] == {
|
||||||
|
"date": "2022-07-15T17:41:31.675698",
|
||||||
|
"expiration": "2022-07-15T17:41:31.675698",
|
||||||
|
"token": "43478d05b35e4781598acd76e33832bb",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def test_delete_new_device_key(tokens):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
assert repo.delete_new_device_key() is None
|
||||||
|
assert "new_device" not in read_json(tokens / "tokens.json")
|
||||||
|
|
||||||
|
|
||||||
|
def test_delete_new_device_key_when_empty(empty_keys):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
repo.delete_new_device_key()
|
||||||
|
assert "new_device" not in read_json(empty_keys / "empty_keys.json")
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_invalid_mnemonic_new_device_key(
|
||||||
|
tokens, mock_new_device_key_generate, datadir, mock_token_generate
|
||||||
|
):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
with pytest.raises(InvalidMnemonic):
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_new_device_key(
|
||||||
|
device_name="imnew",
|
||||||
|
mnemonic_phrase="oh-no",
|
||||||
|
)
|
||||||
|
is None
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_not_exists_mnemonic_new_device_key(
|
||||||
|
tokens, mock_new_device_key_generate, mock_token_generate
|
||||||
|
):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
with pytest.raises(NewDeviceKeyNotFound):
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_new_device_key(
|
||||||
|
device_name="imnew",
|
||||||
|
mnemonic_phrase="uniform clarify napkin bid dress search input armor police cross salon because myself uphold slice bamboo hungry park",
|
||||||
|
)
|
||||||
|
is None
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_mnemonic_new_device_key(
|
||||||
|
tokens, mock_new_device_key_generate, mock_token_generate
|
||||||
|
):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_new_device_key(
|
||||||
|
device_name="imnew",
|
||||||
|
mnemonic_phrase="captain ribbon toddler settle symbol minute step broccoli bless universe divide bulb",
|
||||||
|
)
|
||||||
|
is not None
|
||||||
|
)
|
||||||
|
# assert read_json(datadir / "tokens.json")["new_device"] == []
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_mnemonic_new_device_key_when_empty(empty_keys):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
with pytest.raises(NewDeviceKeyNotFound):
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_new_device_key(
|
||||||
|
device_name="imnew",
|
||||||
|
mnemonic_phrase="captain ribbon toddler settle symbol minute step broccoli bless universe divide bulb",
|
||||||
|
)
|
||||||
|
is None
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_use_mnemonic_new_device_key_when_null(null_keys):
|
||||||
|
repo = JsonTokensRepository()
|
||||||
|
|
||||||
|
with pytest.raises(NewDeviceKeyNotFound):
|
||||||
|
assert (
|
||||||
|
repo.use_mnemonic_new_device_key(
|
||||||
|
device_name="imnew",
|
||||||
|
mnemonic_phrase="captain ribbon toddler settle symbol minute step broccoli bless universe divide bulb",
|
||||||
|
)
|
||||||
|
is None
|
||||||
|
)
|
||||||
|
|
|
@ -0,0 +1,9 @@
|
||||||
|
{
|
||||||
|
"tokens": [
|
||||||
|
{
|
||||||
|
"token": "KG9ni-B-CMPk327Zv1qC7YBQaUGaBUcgdkvMvQ2atFI",
|
||||||
|
"name": "primary_token",
|
||||||
|
"date": "2022-07-15 17:41:31.675698"
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
|
@ -0,0 +1,26 @@
|
||||||
|
{
|
||||||
|
"tokens": [
|
||||||
|
{
|
||||||
|
"token": "KG9ni-B-CMPk327Zv1qC7YBQaUGaBUcgdkvMvQ2atFI",
|
||||||
|
"name": "primary_token",
|
||||||
|
"date": "2022-07-15 17:41:31.675698"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"token": "3JKgLOtFu6ZHgE4OU-R-VdW47IKpg-YQL0c6n7bol68",
|
||||||
|
"name": "second_token",
|
||||||
|
"date": "2022-07-15 17:41:31.675698Z"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"token": "LYiwFDekvALKTQSjk7vtMQuNP_6wqKuV-9AyMKytI_8",
|
||||||
|
"name": "third_token",
|
||||||
|
"date": "2022-07-15T17:41:31.675698Z"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"token": "dD3CFPcEZvapscgzWb7JZTLog7OMkP7NzJeu2fAazXM",
|
||||||
|
"name": "forth_token",
|
||||||
|
"date": "2022-07-15T17:41:31.675698"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"recovery_token": null,
|
||||||
|
"new_device": null
|
||||||
|
}
|
Loading…
Reference in New Issue