From cd15230a92efb94e5938d84c313a5f76f9c297a5 Mon Sep 17 00:00:00 2001 From: Lex Berezhny Date: Fri, 15 Mar 2019 01:33:41 -0400 Subject: [PATCH] getting close to a nice model --- lbrynet/schema/__init__.py | 14 +- lbrynet/schema/claim.py | 467 ++++++++++++++++++--- lbrynet/schema/compat.py | 68 +++ setup.py | 1 - tests/unit/schema/test_claim_from_bytes.py | 172 ++++++++ tests/unit/schema/test_models.py | 52 +++ 6 files changed, 692 insertions(+), 82 deletions(-) create mode 100644 lbrynet/schema/compat.py create mode 100644 tests/unit/schema/test_claim_from_bytes.py create mode 100644 tests/unit/schema/test_models.py diff --git a/lbrynet/schema/__init__.py b/lbrynet/schema/__init__.py index 6c86a5fbb..b5ab2a718 100644 --- a/lbrynet/schema/__init__.py +++ b/lbrynet/schema/__init__.py @@ -1,13 +1 @@ -import os - -__version__ = "0.0.16" - -BLOCKCHAIN_NAME_ENVVAR = "LBRYSCHEMA_BLOCKCHAIN_NAME" -if BLOCKCHAIN_NAME_ENVVAR in os.environ: - if os.environ[BLOCKCHAIN_NAME_ENVVAR] in ['lbrycrd_main', 'lbrycrd_regtest', - 'lbrycrd_testnet']: - BLOCKCHAIN_NAME = os.environ[BLOCKCHAIN_NAME_ENVVAR] - else: - raise OSError("invalid blockchain name: %s" % os.environ[BLOCKCHAIN_NAME_ENVVAR]) -else: - BLOCKCHAIN_NAME = "lbrycrd_main" +from .claim import Claim \ No newline at end of file diff --git a/lbrynet/schema/claim.py b/lbrynet/schema/claim.py index 6c131a4c2..5a8475d84 100644 --- a/lbrynet/schema/claim.py +++ b/lbrynet/schema/claim.py @@ -1,26 +1,23 @@ import json -import binascii -from copy import deepcopy from collections import OrderedDict +from typing import List, Tuple +from decimal import Decimal +from binascii import hexlify, unhexlify -import google.protobuf.json_format as json_pb # pylint: disable=no-name-in-module from google.protobuf import json_format # pylint: disable=no-name-in-module from google.protobuf.message import DecodeError as DecodeError_pb # pylint: disable=no-name-in-module,import-error -from google.protobuf.message import Message # pylint: disable=no-name-in-module,import-error -from lbrynet.schema.types.v2 import claim_pb2 as claim_pb from torba.client.constants import COIN -from lbrynet.schema.types.v1 import legacy_claim_pb2 from lbrynet.schema.signature import Signature from lbrynet.schema.validator import get_validator from lbrynet.schema.signer import get_signer -from lbrynet.schema.legacy_schema_v1.claim import Claim as LegacyClaim -from lbrynet.schema.legacy_schema_v1 import CLAIM_TYPE_NAMES from lbrynet.schema.constants import CURVE_NAMES, SECP256k1 from lbrynet.schema.encoding import decode_fields, decode_b64_fields, encode_fields from lbrynet.schema.error import DecodeError -from lbrynet.schema.fee import Fee +from lbrynet.schema.types.v2.claim_pb2 import Claim as ClaimMessage, Fee as FeeMessage +from lbrynet.schema.base import b58decode, b58encode +from lbrynet.schema import compat class ClaimDict(OrderedDict): @@ -205,66 +202,400 @@ class ClaimDict(OrderedDict): return get_validator(curve).load_from_certificate(claim, certificate_id) -class Schema(Message): - @classmethod - def load(cls, message): - raise NotImplementedError +class Claim: + + __slots__ = '_claim', + + def __init__(self, claim_message=None): + self._claim = claim_message or ClaimMessage() + + @property + def is_undetermined(self): + return self._claim.WhichOneof('type') is None + + @property + def is_stream(self): + return self._claim.WhichOneof('type') == 'stream' + + @property + def is_channel(self): + return self._claim.WhichOneof('type') == 'channel' + + @property + def stream_message(self): + if self.is_undetermined: + self._claim.stream.SetInParent() + if not self.is_stream: + raise ValueError('Claim is not a stream.') + return self._claim.stream + + @property + def stream(self) -> 'Stream': + return Stream(self) + + @property + def channel_message(self): + if self.is_undetermined: + self._claim.channel.SetInParent() + if not self.is_channel: + raise ValueError('Claim is not a channel.') + return self._claim.channel + + @property + def channel(self) -> 'Channel': + return Channel(self) + + def to_bytes(self) -> bytes: + return self._claim.SerializeToString() @classmethod - def _load(cls, data, message): - if isinstance(data, dict): - data = json.dumps(data) - return json_pb.Parse(data, message) - - -class Claim(Schema): - CLAIM_TYPE_STREAM = 0 #fixme: 0 is unset, should be fixed on proto file to be 1 and 2! - CLAIM_TYPE_CERT = 1 - - @classmethod - def load(cls, message: dict): - _claim = deepcopy(message) - _message_pb = claim_pb.Claim() - - if "certificate" in _claim: # old protobuf, migrate - _cert = _claim.pop("certificate") - assert isinstance(_cert, dict) - _message_pb.type = Claim.CLAIM_TYPE_CERT - _message_pb.channel.MergeFrom(claim_pb.Channel(public_key=_cert.pop("publicKey"))) - _claim = {} # so we dont need to know what other fields we ignored - elif "channel" in _claim: - _channel = _claim.pop("channel") - _message_pb.type = Claim.CLAIM_TYPE_CERT - _message_pb.channel = claim_pb.Channel(**_channel) - elif "stream" in _claim: - _message_pb.type = Claim.CLAIM_TYPE_STREAM - _stream = _claim.pop("stream") - if "source" in _stream: - _source = _stream.pop("source") - _message_pb.stream.hash = _source.get("source", b'') # fixme: fail if empty? - _message_pb.stream.media_type = _source.pop("contentType") - if "metadata" in _stream: - _metadata = _stream.pop("metadata") - _message_pb.stream.license = _metadata.get("license") - _message_pb.stream.description = _metadata.get("description") - _message_pb.stream.language = _metadata.get("language") - _message_pb.stream.title = _metadata.get("title") - _message_pb.stream.author = _metadata.get("author") - _message_pb.stream.license_url = _metadata.get("licenseUrl") - _message_pb.stream.thumbnail_url = _metadata.get("thumbnail") - if _metadata.get("nsfw"): - _message_pb.stream.tags.append("nsfw") - if "fee" in _metadata: - _message_pb.stream.fee.address = _metadata["fee"]["address"] - _message_pb.stream.fee.currency = { - "LBC": 0, - "USD": 1 - }[_metadata["fee"]["currency"]] - multiplier = COIN if _metadata["fee"]["currency"] == "LBC" else 100 - total = int(_metadata["fee"]["amount"]*multiplier) - _message_pb.stream.fee.amount = total if total >= 0 else 0 - _claim = {} + def from_bytes(cls, data: bytes) -> 'Claim': + claim = ClaimMessage() + if data[0] == 0: + claim.ParseFromString(data[1:]) + return cls(claim) + elif data[0] == 1: + claim.ParseFromString(data[85:]) + return cls(claim).from_message(payload[1:21], payload[21:85]) + elif data[0] == ord('{'): + return compat.from_old_json_schema(cls(claim), data) else: - raise AttributeError + return compat.from_types_v1(cls(claim), data) - return cls._load(_claim, _message_pb) + +class Video: + + __slots__ = '_video', + + def __init__(self, video_message): + self._video = video_message + + @property + def width(self) -> int: + return self._video.width + + @width.setter + def width(self, width: int): + self._video.width = width + + @property + def height(self) -> int: + return self._video.height + + @height.setter + def height(self, height: int): + self._video.height = height + + @property + def dimensions(self) -> Tuple[int, int]: + return self.width, self.height + + @dimensions.setter + def dimensions(self, dimensions: Tuple[int, int]): + self._video.width, self._video.height = dimensions + + +class File: + + __slots__ = '_file', + + def __init__(self, file_message): + self._file = file_message + + @property + def name(self) -> str: + return self._file.name + + @name.setter + def name(self, name: str): + self._file.name = name + + @property + def size(self) -> int: + return self._file.size + + @size.setter + def size(self, size: int): + self._file.size = size + + +class Fee: + + __slots__ = '_fee', + + def __init__(self, fee_message): + self._fee = fee_message + + @property + def currency(self) -> str: + return FeeMessage.Currency.Name(self._fee.currency) + + @currency.setter + def currency(self, currency: str): + self._fee.currency = FeeMessage.Currency.Value(currency) + + @property + def address(self) -> str: + return b58encode(self._fee.address).decode() + + @address.setter + def address(self, address: str): + self._fee.address = b58decode(address) + + @property + def address_bytes(self) -> bytes: + return self._fee.address + + @address_bytes.setter + def address_bytes(self, address: bytes): + self._fee.address = address + + @property + def dewies(self) -> int: + if self._fee.currency != FeeMessage.LBC: + raise ValueError('Dewies can only be returned for LBC fees.') + return self._fee.amount + + @dewies.setter + def dewies(self, amount: int): + self._fee.amount = amount + self._fee.currency = FeeMessage.LBC + + DEWEYS = Decimal(COIN) + + @property + def lbc(self) -> Decimal: + if self._fee.currency != FeeMessage.LBC: + raise ValueError('LBC can only be returned for LBC fees.') + return Decimal(self._fee.amount / self.DEWEYS) + + @lbc.setter + def lbc(self, amount: Decimal): + self.dewies = int(amount * self.DEWEYS) + + USD = Decimal(100.0) + + @property + def usd(self) -> Decimal: + if self._fee.currency != FeeMessage.USD: + raise ValueError('USD can only be returned for USD fees.') + return Decimal(self._fee.amount / self.USD) + + @usd.setter + def usd(self, amount: Decimal): + self._fee.amount = int(amount * self.USD) + self._fee.currency = FeeMessage.USD + + +class Stream: + + __slots__ = '_claim', '_stream' + + def __init__(self, claim: Claim = None): + self._claim = claim or Claim() + self._stream = self._claim.stream_message + + @property + def claim(self) -> Claim: + return self._claim + + @property + def video(self) -> Video: + return Video(self._stream.video) + + @property + def file(self) -> File: + return File(self._stream.file) + + @property + def fee(self) -> Fee: + return Fee(self._stream.fee) + + @property + def tags(self) -> List: + return self._stream.tags + + @property + def hash(self) -> str: + return hexlify(self._stream.hash).decode() + + @hash.setter + def hash(self, sd_hash: str): + self._stream.hash = unhexlify(sd_hash.encode()) + + @property + def hash_bytes(self) -> bytes: + return self._stream.hash + + @hash_bytes.setter + def hash_bytes(self, hash: bytes): + self._stream.hash = hash + + @property + def language(self) -> str: + return self._stream.language + + @language.setter + def language(self, language: str): + self._stream.language = language + + @property + def title(self) -> str: + return self._stream.title + + @title.setter + def title(self, title: str): + self._stream.title = title + + @property + def author(self) -> str: + return self._stream.author + + @author.setter + def author(self, author: str): + self._stream.author = author + + @property + def description(self) -> str: + return self._stream.description + + @description.setter + def description(self, description: str): + self._stream.description = description + + @property + def media_type(self) -> str: + return self._stream.media_type + + @media_type.setter + def media_type(self, media_type: str): + self._stream.media_type = media_type + + @property + def license(self) -> str: + return self._stream.license + + @license.setter + def license(self, license: str): + self._stream.license = license + + @property + def license_url(self) -> str: + return self._stream.license_url + + @license_url.setter + def license_url(self, license_url: str): + self._stream.license_url = license_url + + @property + def thumbnail_url(self) -> str: + return self._stream.thumbnail_url + + @thumbnail_url.setter + def thumbnail_url(self, thumbnail_url: str): + self._stream.thumbnail_url = thumbnail_url + + @property + def duration(self) -> int: + return self._stream.duration + + @duration.setter + def duration(self, duration: int): + self._stream.duration = duration + + @property + def release_time(self) -> int: + return self._stream.release_time + + @release_time.setter + def release_time(self, release_time: int): + self._stream.release_time = release_time + + +class Channel: + + __slots__ = '_claim', '_channel' + + def __init__(self, claim: Claim = None): + self._claim = claim or Claim() + self._channel = self._claim.channel_message + + @property + def claim(self) -> Claim: + return self._claim + + @property + def tags(self) -> List: + return self._channel.tags + + @property + def public_key(self) -> str: + return hexlify(self._channel.public_key).decode() + + @public_key.setter + def public_key(self, sd_public_key: str): + self._channel.public_key = unhexlify(sd_public_key.encode()) + + @property + def public_key_bytes(self) -> bytes: + return self._channel.public_key + + @public_key_bytes.setter + def public_key_bytes(self, public_key: bytes): + self._channel.public_key = public_key + + @property + def language(self) -> str: + return self._channel.language + + @language.setter + def language(self, language: str): + self._channel.language = language + + @property + def title(self) -> str: + return self._channel.title + + @title.setter + def title(self, title: str): + self._channel.title = title + + @property + def description(self) -> str: + return self._channel.description + + @description.setter + def description(self, description: str): + self._channel.description = description + + @property + def contact_email(self) -> str: + return self._channel.contact_email + + @contact_email.setter + def contact_email(self, contact_email: str): + self._channel.contact_email = contact_email + + @property + def homepage_url(self) -> str: + return self._channel.homepage_url + + @homepage_url.setter + def homepage_url(self, homepage_url: str): + self._channel.homepage_url = homepage_url + + @property + def thumbnail_url(self) -> str: + return self._channel.thumbnail_url + + @thumbnail_url.setter + def thumbnail_url(self, thumbnail_url: str): + self._channel.thumbnail_url = thumbnail_url + + @property + def cover_url(self) -> str: + return self._channel.cover_url + + @cover_url.setter + def cover_url(self, cover_url: str): + self._channel.cover_url = cover_url diff --git a/lbrynet/schema/compat.py b/lbrynet/schema/compat.py new file mode 100644 index 000000000..046eb0064 --- /dev/null +++ b/lbrynet/schema/compat.py @@ -0,0 +1,68 @@ +import json +from decimal import Decimal + +from lbrynet.schema.address import decode_address, encode_address +from lbrynet.schema.types.v1.legacy_claim_pb2 import Claim as OldClaimMessage +from lbrynet.schema.types.v1.metadata_pb2 import Metadata as MetadataMessage +from lbrynet.schema.types.v1.fee_pb2 import Fee as FeeMessage + + +def from_old_json_schema(claim, payload: bytes): + value = json.loads(payload) + stream = claim.stream + stream.media_type = value.get('content_type', value.get('content-type', 'application/octet-stream')) + stream.title = value.get('title', '') + stream.description = value.get('description', '') + stream.thumbnail_url = value.get('thumbnail', '') + stream.author = value.get('author', '') + stream.license = value.get('license', '') + stream.license_url = value.get('license_url', '') + stream.language = value.get('language', '') + stream.hash = value['sources']['lbry_sd_hash'] + if value.get('nsfw', False): + stream.tags.append('nsfw') + if "fee" in value: + fee = value["fee"] + currency = list(fee.keys())[0] + if currency == 'LBC': + stream.fee.lbc = Decimal(fee[currency]['amount']) + elif currency == 'USD': + stream.fee.usd = Decimal(fee[currency]['amount']) + else: + raise ValueError(f'Unknown currency: {currency}') + stream.fee.address = fee[currency]['address'] + return claim + + +def from_types_v1(claim, payload: bytes): + old = OldClaimMessage() + old.ParseFromString(payload) + if old.claimType == 1: + stream = claim.stream + stream.title = old.stream.metadata.title + stream.description = old.stream.metadata.description + stream.author = old.stream.metadata.author + stream.license = old.stream.metadata.license + stream.license_url = old.stream.metadata.licenseUrl + stream.thumbnail_url = old.stream.metadata.thumbnail + stream.language = MetadataMessage.Language.Name(old.stream.metadata.language) + stream.media_type = old.stream.source.contentType + stream.hash_bytes = old.stream.source.source + if old.stream.metadata.nsfw: + stream.tags.append('nsfw') + if old.stream.metadata.HasField('fee'): + fee = old.stream.metadata.fee + stream.fee.address_bytes = fee.address + currency = FeeMessage.Currency.Name(fee.currency) + if currency == 'LBC': + stream.fee.lbc = Decimal(fee.amount) + elif currency == 'USD': + stream.fee.usd = Decimal(fee.amount) + else: + raise ValueError(f'Unsupported currency: {currency}') + elif old.claimType == 2: + channel = claim.channel + channel.public_key_bytes = old.certificate.publicKey + else: + raise ValueError('claimType must be 1 for Streams and 2 for Channel') + return claim diff --git a/setup.py b/setup.py index 27df21ce7..d581c58bf 100644 --- a/setup.py +++ b/setup.py @@ -34,7 +34,6 @@ setup( 'cryptography==2.5', 'protobuf==3.6.1', 'msgpack==0.6.1', - 'jsonschema==2.6.0', 'ecdsa==0.13', 'torba', 'pyyaml==3.13', diff --git a/tests/unit/schema/test_claim_from_bytes.py b/tests/unit/schema/test_claim_from_bytes.py new file mode 100644 index 000000000..37aef4f8d --- /dev/null +++ b/tests/unit/schema/test_claim_from_bytes.py @@ -0,0 +1,172 @@ +from unittest import TestCase +from binascii import unhexlify + +from lbrynet.schema import Claim + + +class TestOldJSONSchemaCompatibility(TestCase): + + def test_old_json_schema_v1(self): + claim = Claim.from_bytes( + b'{"fee": {"LBC": {"amount": 1.0, "address": "bPwGA9h7uijoy5uAvzVPQw9QyLoYZehHJo"}}, "d' + b'escription": "10MB test file to measure download speed on Lbry p2p-network.", "licens' + b'e": "None", "author": "root", "language": "English", "title": "10MB speed test file",' + b' "sources": {"lbry_sd_hash": "bbd1f68374ff9a1044a90d7dd578ce41979211c386caf19e6f49653' + b'6db5f2c96b58fe2c7a6677b331419a117873b539f"}, "content-type": "application/octet-strea' + b'm", "thumbnail": "/home/robert/lbry/speed.jpg"}' + ) + stream = claim.stream + self.assertEqual(stream.title, '10MB speed test file') + self.assertEqual(stream.description, '10MB test file to measure download speed on Lbry p2p-network.') + self.assertEqual(stream.license, 'None') + self.assertEqual(stream.author, 'root') + self.assertEqual(stream.language, 'English') + self.assertEqual(stream.media_type, 'application/octet-stream') + self.assertEqual(stream.thumbnail_url, '/home/robert/lbry/speed.jpg') + self.assertEqual( + stream.hash, + 'bbd1f68374ff9a1044a90d7dd578ce41979211c386caf19e' + '6f496536db5f2c96b58fe2c7a6677b331419a117873b539f' + ) + self.assertEqual(stream.fee.address, 'bPwGA9h7uijoy5uAvzVPQw9QyLoYZehHJo') + self.assertEqual(stream.fee.lbc, 1) + self.assertEqual(stream.fee.dewies, 100000000) + self.assertEqual(stream.fee.currency, 'LBC') + with self.assertRaisesRegex(ValueError, 'USD can only be returned for USD fees.'): + print(stream.fee.usd) + + def test_old_json_schema_v2(self): + claim = Claim.from_bytes( + b'{"license": "Creative Commons Attribution 3.0 United States", "fee": {"LBC": {"amount' + b'": 10, "address": "bFro33qBKxnL1AsjUU9N4AQHp9V62Nhc5L"}}, "ver": "0.0.2", "descriptio' + b'n": "Force P0 State for Nividia Cards! (max mining performance)", "language": "en", "' + b'author": "Mii", "title": "Nividia P0", "sources": {"lbry_sd_hash": "c5ffee0fa5168e166' + b'81b519d9d85446e8d1d818a616bd55540aa7374d2321b51abf2ac3dae1443a03dadcc8f7affaa62"}, "n' + b'sfw": false, "license_url": "https://creativecommons.org/licenses/by/3.0/us/legalcode' + b'", "content-type": "application/x-msdownload"}' + ) + stream = claim.stream + self.assertEqual(stream.title, 'Nividia P0') + self.assertEqual(stream.description, 'Force P0 State for Nividia Cards! (max mining performance)') + self.assertEqual(stream.license, 'Creative Commons Attribution 3.0 United States') + self.assertEqual(stream.license_url, 'https://creativecommons.org/licenses/by/3.0/us/legalcode') + self.assertEqual(stream.author, 'Mii') + self.assertEqual(stream.language, 'en') + self.assertEqual(stream.media_type, 'application/x-msdownload') + self.assertEqual( + stream.hash, + 'c5ffee0fa5168e16681b519d9d85446e8d1d818a616bd555' + '40aa7374d2321b51abf2ac3dae1443a03dadcc8f7affaa62' + ) + self.assertEqual(stream.fee.address, 'bFro33qBKxnL1AsjUU9N4AQHp9V62Nhc5L') + self.assertEqual(stream.fee.lbc, 10) + self.assertEqual(stream.fee.dewies, 1000000000) + self.assertEqual(stream.fee.currency, 'LBC') + with self.assertRaisesRegex(ValueError, 'USD can only be returned for USD fees.'): + print(stream.fee.usd) + + def test_old_json_schema_v3(self): + claim = Claim.from_bytes( + b'{"ver": "0.0.3", "description": "asd", "license": "Creative Commons Attribution 4.0 I' + b'nternational", "author": "sgb", "title": "ads", "language": "en", "sources": {"lbry_s' + b'd_hash": "d83db664c6d7d570aa824300f4869e0bfb560e765efa477aebf566467f8d3a57f4f8c704cab' + b'1308eb75ff8b7e84e3caf"}, "content_type": "video/mp4", "nsfw": false}' + ) + stream = claim.stream + self.assertEqual(stream.title, 'ads') + self.assertEqual(stream.description, 'asd') + self.assertEqual(stream.license, 'Creative Commons Attribution 4.0 International') + self.assertEqual(stream.author, 'sgb') + self.assertEqual(stream.language, 'en') + self.assertEqual(stream.media_type, 'video/mp4') + self.assertEqual( + stream.hash, + 'd83db664c6d7d570aa824300f4869e0bfb560e765efa477a' + 'ebf566467f8d3a57f4f8c704cab1308eb75ff8b7e84e3caf' + ) + + +class TestTypesV1Compatibility(TestCase): + + def test_signed_claim_made_by_ytsync(self): + claim = Claim.from_bytes(unhexlify( + b'080110011aee04080112a604080410011a2b4865726520617265203520526561736f6e73204920e29da4e' + b'fb88f204e657874636c6f7564207c20544c4722920346696e64206f7574206d6f72652061626f7574204e' + b'657874636c6f75643a2068747470733a2f2f6e657874636c6f75642e636f6d2f0a0a596f752063616e206' + b'6696e64206d65206f6e20746865736520736f6369616c733a0a202a20466f72756d733a2068747470733a' + b'2f2f666f72756d2e6865617679656c656d656e742e696f2f0a202a20506f64636173743a2068747470733' + b'a2f2f6f6666746f706963616c2e6e65740a202a2050617472656f6e3a2068747470733a2f2f7061747265' + b'6f6e2e636f6d2f7468656c696e757867616d65720a202a204d657263683a2068747470733a2f2f7465657' + b'37072696e672e636f6d2f73746f7265732f6f6666696369616c2d6c696e75782d67616d65720a202a2054' + b'77697463683a2068747470733a2f2f7477697463682e74762f786f6e64616b0a202a20547769747465723' + b'a2068747470733a2f2f747769747465722e636f6d2f7468656c696e757867616d65720a0a2e2e2e0a6874' + b'7470733a2f2f7777772e796f75747562652e636f6d2f77617463683f763d4672546442434f535f66632a0' + b'f546865204c696e75782047616d6572321c436f7079726967687465642028636f6e746163742061757468' + b'6f722938004a2968747470733a2f2f6265726b2e6e696e6a612f7468756d626e61696c732f46725464424' + b'34f535f666352005a001a41080110011a30040e8ac6e89c061f982528c23ad33829fd7146435bf7a4cc22' + b'f0bff70c4fe0b91fd36da9a375e3e1c171db825bf5d1f32209766964656f2f6d70342a5c080110031a406' + b'2b2dd4c45e364030fbfad1a6fefff695ebf20ea33a5381b947753e2a0ca359989a5cc7d15e5392a0d354c' + b'0b68498382b2701b22c03beb8dcb91089031b871e72214feb61536c007cdf4faeeaab4876cb397feaf6b51' + )) + stream = claim.stream + self.assertEqual(stream.title, 'Here are 5 Reasons I ❤️ Nextcloud | TLG') + self.assertEqual( + stream.description, + 'Find out more about Nextcloud: https://nextcloud.com/\n\nYou can find me on these soci' + 'als:\n * Forums: https://forum.heavyelement.io/\n * Podcast: https://offtopical.net\n ' + '* Patreon: https://patreon.com/thelinuxgamer\n * Merch: https://teespring.com/stores/o' + 'fficial-linux-gamer\n * Twitch: https://twitch.tv/xondak\n * Twitter: https://twitter.' + 'com/thelinuxgamer\n\n...\nhttps://www.youtube.com/watch?v=FrTdBCOS_fc' + ) + self.assertEqual(stream.license, 'Copyrighted (contact author)') + self.assertEqual(stream.author, 'The Linux Gamer') + self.assertEqual(stream.language, 'en') + self.assertEqual(stream.media_type, 'video/mp4') + self.assertEqual(stream.thumbnail_url, 'https://berk.ninja/thumbnails/FrTdBCOS_fc') + self.assertEqual( + stream.hash, + '040e8ac6e89c061f982528c23ad33829fd7146435bf7a4cc' + '22f0bff70c4fe0b91fd36da9a375e3e1c171db825bf5d1f3' + ) + + # certificate for above channel + cert = Claim.from_bytes(unhexlify( + b'08011002225e0801100322583056301006072a8648ce3d020106052b8104000a034200043878b1edd4a13' + b'73149909ef03f4339f6da9c2bd2214c040fd2e530463ffe66098eca14fc70b50ff3aefd106049a815f595' + b'ed5a13eda7419ad78d9ed7ae473f17' + )) + channel = cert.channel + self.assertEqual( + channel.public_key, + '3056301006072a8648ce3d020106052b8104000a034200043878b1edd4a1373149909ef03f4339f6da9c2b' + 'd2214c040fd2e530463ffe66098eca14fc70b50ff3aefd106049a815f595ed5a13eda7419ad78d9ed7ae47' + '3f17' + ) + + def test_unsigned_with_fee(self): + claim = Claim.from_bytes(unhexlify( + b'080110011ad6010801127c080410011a08727067206d69646922046d6964692a08727067206d696469322' + b'e437265617469766520436f6d6d6f6e73204174747269627574696f6e20342e3020496e7465726e617469' + b'6f6e616c38004224080110011a19553f00bc139bbf40de425f94d51fffb34c1bea6d9171cd374c2500007' + b'0414a0052005a001a54080110011a301f41eb0312aa7e8a5ce49349bc77d811da975833719d751523b19f' + b'123fc3d528d6a94e3446ccddb7b9329f27a9cad7e3221c6170706c69636174696f6e2f782d7a69702d636' + b'f6d70726573736564' + )) + stream = claim.stream + self.assertEqual(stream.title, 'rpg midi') + self.assertEqual(stream.description, 'midi') + self.assertEqual(stream.license, 'Creative Commons Attribution 4.0 International') + self.assertEqual(stream.author, 'rpg midi') + self.assertEqual(stream.language, 'en') + self.assertEqual(stream.media_type, 'application/x-zip-compressed') + self.assertEqual( + stream.hash, + '1f41eb0312aa7e8a5ce49349bc77d811da975833719d7515' + '23b19f123fc3d528d6a94e3446ccddb7b9329f27a9cad7e3' + ) + self.assertEqual(stream.fee.address, 'bJUQ9MxS9N6M29zsA5GTpVSDzsnPjMBBX9') + self.assertEqual(stream.fee.lbc, 15) + self.assertEqual(stream.fee.dewies, 1500000000) + self.assertEqual(stream.fee.currency, 'LBC') + with self.assertRaisesRegex(ValueError, 'USD can only be returned for USD fees.'): + print(stream.fee.usd) diff --git a/tests/unit/schema/test_models.py b/tests/unit/schema/test_models.py new file mode 100644 index 000000000..9d6d96c72 --- /dev/null +++ b/tests/unit/schema/test_models.py @@ -0,0 +1,52 @@ +from unittest import TestCase +from decimal import Decimal + +from lbrynet.schema.claim import Claim, Channel, Stream + + +class TestClaimContainerAwareness(TestCase): + + def test_stream_claim(self): + stream = Stream() + claim = stream.claim + self.assertTrue(claim.is_stream) + self.assertFalse(claim.is_channel) + claim = Claim.from_bytes(claim.to_bytes()) + self.assertTrue(claim.is_stream) + self.assertFalse(claim.is_channel) + self.assertIsNotNone(claim.stream) + with self.assertRaisesRegex(ValueError, 'Claim is not a channel.'): + print(claim.channel) + + def test_channel_claim(self): + channel = Channel() + claim = channel.claim + self.assertFalse(claim.is_stream) + self.assertTrue(claim.is_channel) + claim = Claim.from_bytes(claim.to_bytes()) + self.assertFalse(claim.is_stream) + self.assertTrue(claim.is_channel) + self.assertIsNotNone(claim.channel) + with self.assertRaisesRegex(ValueError, 'Claim is not a stream.'): + print(claim.stream) + + +class TestFee(TestCase): + + def test_amount_setters(self): + stream = Stream() + + stream.fee.lbc = Decimal('1.01') + self.assertEqual(stream.fee.lbc, Decimal('1.01')) + self.assertEqual(stream.fee.dewies, 101000000) + self.assertEqual(stream.fee.currency, 'LBC') + with self.assertRaisesRegex(ValueError, 'USD can only be returned for USD fees.'): + print(stream.fee.usd) + + stream.fee.usd = Decimal('1.01') + self.assertEqual(stream.fee.usd, Decimal('1.01')) + self.assertEqual(stream.fee.currency, 'USD') + with self.assertRaisesRegex(ValueError, 'LBC can only be returned for LBC fees.'): + print(stream.fee.lbc) + with self.assertRaisesRegex(ValueError, 'Dewies can only be returned for LBC fees.'): + print(stream.fee.dewies)