Spaces:
Sleeping
Sleeping
File size: 2,019 Bytes
7864524 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | import json
import pytest
from app.services.token_dao import TokenDAO
from app.services.token_importer import import_tokens_from_directory
@pytest.mark.asyncio
async def test_import_tokens_from_directory_handles_duplicates_and_invalid_files(
tmp_path,
):
source_dir = tmp_path / "source_tokens"
source_dir.mkdir()
(source_dir / "token_valid_1.json").write_text(
json.dumps(
{
"email": "alpha@example.com",
"token": "token-alpha",
"token_source": "context.cookie:token",
}
),
encoding="utf-8",
)
(source_dir / "token_valid_2.json").write_text(
json.dumps(
{
"email": "beta@example.com",
"token": "token-beta",
"token_source": "context.cookie:token",
}
),
encoding="utf-8",
)
(source_dir / "token_duplicate.json").write_text(
json.dumps(
{
"email": "alpha-dup@example.com",
"token": "token-alpha",
}
),
encoding="utf-8",
)
(source_dir / "token_missing.json").write_text(
json.dumps({"email": "missing@example.com"}),
encoding="utf-8",
)
(source_dir / "token_invalid.json").write_text("{invalid json", encoding="utf-8")
dao = TokenDAO(str(tmp_path / "tokens.db"))
await dao.init_database()
summary = await import_tokens_from_directory(
source_dir,
provider="zai",
validate=False,
dao=dao,
)
assert summary.scanned_files == 5
assert summary.imported_count == 2
assert summary.duplicate_count == 1
assert summary.missing_token_count == 1
assert summary.invalid_json_count == 1
assert summary.invalid_token_count == 0
tokens = await dao.get_tokens_by_provider("zai", enabled_only=False)
imported_values = {item["token"] for item in tokens}
assert imported_values == {"token-alpha", "token-beta"}
|