133 lines
6.1 KiB
Python
133 lines
6.1 KiB
Python
import base64
|
||
import os
|
||
from typing import Any, Dict
|
||
|
||
import pytest
|
||
|
||
pytestmark = pytest.mark.ton
|
||
|
||
|
||
try:
|
||
from app.core._blockchain.ton.nft_license_manager import NFTLicenseManager # высокоуровневый менеджер TON NFT
|
||
except Exception:
|
||
NFTLicenseManager = None # type: ignore
|
||
|
||
try:
|
||
from app.core.access.content_access_manager import ContentAccessManager
|
||
except Exception:
|
||
ContentAccessManager = None # type: ignore
|
||
|
||
try:
|
||
from app.core.models.license.nft_license import NFTLicense
|
||
except Exception:
|
||
NFTLicense = None # type: ignore
|
||
|
||
|
||
class _MockTonBackend:
|
||
"""
|
||
Минимальный мок backend TON для изоляции тестов:
|
||
- issue_nft(content_id, owner) -> {"nft_address": "...", "tx_hash": "..."}
|
||
- verify_ownership(nft_address, owner) -> bool
|
||
"""
|
||
def __init__(self) -> None:
|
||
self._owners: Dict[str, str] = {}
|
||
|
||
def issue_nft(self, content_id: str, owner: str) -> Dict[str, str]:
|
||
addr = "EQ" + os.urandom(20).hex()
|
||
self._owners[addr] = owner
|
||
return {"nft_address": addr, "tx_hash": os.urandom(16).hex()}
|
||
|
||
def verify_ownership(self, nft_address: str, owner: str) -> bool:
|
||
return self._owners.get(nft_address) == owner
|
||
|
||
|
||
@pytest.mark.skipif(NFTLicenseManager is None or NFTLicense is None, reason="TON/NFT components not importable")
|
||
def test_nft_issue_and_verify_access_with_mock():
|
||
"""
|
||
Тестируем логику выдачи и проверки доступа через NFT лицензию на уровне менеджера,
|
||
изолируя внешние сетевые вызовы.
|
||
"""
|
||
backend = _MockTonBackend()
|
||
# Инициализация менеджера: если у менеджера другой конструктор — этот тест подскажет адаптацию.
|
||
try:
|
||
mgr = NFTLicenseManager(backend=backend) # type: ignore[call-arg]
|
||
except TypeError:
|
||
# Фоллбек: если менеджер не принимает backend, подменим методы через monkeypatch в другом тесте
|
||
pytest.skip("NFTLicenseManager doesn't support DI for backend; adapt test to your implementation")
|
||
|
||
owner = "EQ_OWNER_001"
|
||
content_id = "CID-" + os.urandom(4).hex()
|
||
|
||
res = backend.issue_nft(content_id, owner)
|
||
nft_addr = res["nft_address"]
|
||
|
||
lic = NFTLicense(
|
||
license_id="LIC-" + os.urandom(3).hex(),
|
||
content_id=content_id,
|
||
owner_address=owner,
|
||
nft_address=nft_addr,
|
||
)
|
||
|
||
assert backend.verify_ownership(lic.nft_address, owner), "Ownership must be verified by mock backend"
|
||
# Если у менеджера есть валидация, используем ее
|
||
ver_ok = True
|
||
if hasattr(mgr, "verify_license"):
|
||
ver_ok = bool(mgr.verify_license(lic.to_dict())) # type: ignore[attr-defined]
|
||
assert ver_ok, "Manager verify_license should accept valid license"
|
||
|
||
|
||
@pytest.mark.skipif(ContentAccessManager is None or NFTLicense is None, reason="Access manager or NFT model not importable")
|
||
def test_access_manager_allows_with_valid_license(monkeypatch):
|
||
"""
|
||
Имитация проверки доступа через ContentAccessManager:
|
||
- Успешный доступ, если есть действующая NFT лицензия и владелец совпадает.
|
||
"""
|
||
cam = ContentAccessManager() # type: ignore[call-arg]
|
||
owner = "EQ_OWNER_002"
|
||
content_id = "CID-" + os.urandom(4).hex()
|
||
lic = NFTLicense(
|
||
license_id="LIC-OK",
|
||
content_id=content_id,
|
||
owner_address=owner,
|
||
nft_address="EQ_FAKE_NFT_ADDR",
|
||
)
|
||
|
||
# Подменим методы, чтобы Cam считал лицензию валидной
|
||
if hasattr(cam, "get_license_by_id"):
|
||
monkeypatch.setattr(cam, "get_license_by_id", lambda _lic_id: lic)
|
||
if hasattr(cam, "is_license_valid_for_owner"):
|
||
monkeypatch.setattr(cam, "is_license_valid_for_owner", lambda l, o: l.owner_address == o)
|
||
|
||
# Унифицированный метод проверки доступа (имя может отличаться в реализации)
|
||
check = getattr(cam, "can_access_content", None)
|
||
if callable(check):
|
||
assert check(license_id=lic.license_id, content_id=content_id, owner_address=owner), "Expected access granted"
|
||
else:
|
||
# Если API иное, используем общие строительные блоки:
|
||
got = cam.get_license_by_id(lic.license_id) if hasattr(cam, "get_license_by_id") else lic # type: ignore[attr-defined]
|
||
valid = cam.is_license_valid_for_owner(got, owner) if hasattr(cam, "is_license_valid_for_owner") else (got.owner_address == owner) # type: ignore[attr-defined]
|
||
assert valid and got.content_id == content_id, "Access validation failed by building blocks"
|
||
|
||
|
||
@pytest.mark.skipif(ContentAccessManager is None or NFTLicense is None, reason="Access manager or NFT model not importable")
|
||
def test_access_manager_denies_on_owner_mismatch(monkeypatch):
|
||
cam = ContentAccessManager() # type: ignore[call-arg]
|
||
content_id = "CID-" + os.urandom(4).hex()
|
||
lic = NFTLicense(
|
||
license_id="LIC-NO",
|
||
content_id=content_id,
|
||
owner_address="EQ_REAL_OWNER",
|
||
nft_address="EQ_FAKE",
|
||
)
|
||
if hasattr(cam, "get_license_by_id"):
|
||
monkeypatch.setattr(cam, "get_license_by_id", lambda _lic_id: lic)
|
||
if hasattr(cam, "is_license_valid_for_owner"):
|
||
monkeypatch.setattr(cam, "is_license_valid_for_owner", lambda l, o: l.owner_address == o)
|
||
|
||
check = getattr(cam, "can_access_content", None)
|
||
if callable(check):
|
||
assert not check(license_id=lic.license_id, content_id=content_id, owner_address="EQ_SOMEONE"), "Access must be denied"
|
||
else:
|
||
got = cam.get_license_by_id(lic.license_id) if hasattr(cam, "get_license_by_id") else lic # type: ignore[attr-defined]
|
||
valid = cam.is_license_valid_for_owner(got, "EQ_SOMEONE") if hasattr(cam, "is_license_valid_for_owner") else (got.owner_address == "EQ_SOMEONE") # type: ignore[attr-defined]
|
||
assert not (valid and got.content_id == content_id), "Access must be denied when owner mismatched" |