125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models import Credential
|
|
from backend.app.services.credential_crypto import CredentialCrypto
|
|
from backend.app.services.credential_service import (
|
|
CredentialService,
|
|
EmptyDataError,
|
|
UnknownPlatformError,
|
|
WrongAuthTypeError,
|
|
)
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
_NETSCAPE = (
|
|
"# Netscape HTTP Cookie File\n"
|
|
".patreon.com\tTRUE\t/\tTRUE\t1700000000\tsession_id\tabc\n"
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def crypto(tmp_path):
|
|
return CredentialCrypto(tmp_path / "k")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_creates_then_updates(db, crypto):
|
|
svc = CredentialService(db, crypto)
|
|
rec = await svc.upsert(
|
|
platform="patreon", credential_type="cookies", data=_NETSCAPE,
|
|
)
|
|
assert rec.platform == "patreon"
|
|
assert rec.credential_type == "cookies"
|
|
again = await svc.upsert(
|
|
platform="patreon", credential_type="cookies", data=_NETSCAPE + "x",
|
|
)
|
|
assert again.platform == "patreon"
|
|
rows = (await db.execute(
|
|
select(Credential).where(Credential.platform == "patreon")
|
|
)).scalars().all()
|
|
assert len(rows) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_rejects_unknown_platform(db, crypto):
|
|
svc = CredentialService(db, crypto)
|
|
with pytest.raises(UnknownPlatformError):
|
|
await svc.upsert(platform="fanbox", credential_type="cookies", data="x")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_rejects_wrong_auth_type(db, crypto):
|
|
svc = CredentialService(db, crypto)
|
|
# discord is token-only
|
|
with pytest.raises(WrongAuthTypeError):
|
|
await svc.upsert(platform="discord", credential_type="cookies", data="x")
|
|
# patreon is cookies-only
|
|
with pytest.raises(WrongAuthTypeError):
|
|
await svc.upsert(platform="patreon", credential_type="token", data="x")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upsert_rejects_empty_data(db, crypto):
|
|
svc = CredentialService(db, crypto)
|
|
with pytest.raises(EmptyDataError):
|
|
await svc.upsert(platform="patreon", credential_type="cookies", data=" ")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_excludes_blob(db, crypto):
|
|
svc = CredentialService(db, crypto)
|
|
await svc.upsert(platform="patreon", credential_type="cookies", data=_NETSCAPE)
|
|
records = await svc.list()
|
|
assert len(records) == 1
|
|
d = records[0].to_dict()
|
|
assert "data" not in d
|
|
assert "encrypted_blob" not in d
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete(db, crypto):
|
|
svc = CredentialService(db, crypto)
|
|
await svc.upsert(platform="patreon", credential_type="cookies", data=_NETSCAPE)
|
|
await svc.delete("patreon")
|
|
assert await svc.get("patreon") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_cookies_path_writes_usable_file(db, crypto, tmp_path):
|
|
svc = CredentialService(db, crypto, cookies_dir=tmp_path / "cookies")
|
|
await svc.upsert(platform="patreon", credential_type="cookies", data=_NETSCAPE)
|
|
path = await svc.get_cookies_path("patreon")
|
|
assert path is not None and path.exists()
|
|
contents = path.read_text()
|
|
assert ".patreon.com" in contents
|
|
assert "session_id" in contents
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_cookies_path_none_for_missing(db, crypto, tmp_path):
|
|
svc = CredentialService(db, crypto, cookies_dir=tmp_path)
|
|
assert await svc.get_cookies_path("patreon") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_cookies_path_none_for_token_kind(db, crypto, tmp_path):
|
|
svc = CredentialService(db, crypto, cookies_dir=tmp_path)
|
|
await svc.upsert(platform="discord", credential_type="token", data="tok")
|
|
assert await svc.get_cookies_path("discord") is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_token_decrypts(db, crypto):
|
|
svc = CredentialService(db, crypto)
|
|
await svc.upsert(platform="discord", credential_type="token", data="my-token")
|
|
assert await svc.get_token("discord") == "my-token"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_token_none_for_cookies_kind(db, crypto):
|
|
svc = CredentialService(db, crypto)
|
|
await svc.upsert(platform="patreon", credential_type="cookies", data=_NETSCAPE)
|
|
assert await svc.get_token("patreon") is None
|