from unittest import TestCase from decimal import Decimal from lbry.schema.claim import Claim, Stream, Collection from lbry.error import InputValueIsNoneError class TestClaimContainerAwareness(TestCase): def test_stream_claim(self): stream = Stream() claim = stream.claim self.assertEqual(claim.claim_type, Claim.STREAM) claim = Claim.from_bytes(claim.to_bytes()) self.assertEqual(claim.claim_type, Claim.STREAM) self.assertIsNotNone(claim.stream) with self.assertRaisesRegex(ValueError, 'Claim is not a channel.'): print(claim.channel) 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') stream.fee.dewies = 203000000 self.assertEqual(stream.fee.lbc, Decimal('2.03')) self.assertEqual(stream.fee.dewies, 203000000) self.assertEqual(stream.fee.currency, 'LBC') with self.assertRaisesRegex(ValueError, 'USD can only be returned for USD fees.'): print(stream.fee.usd) with self.assertRaisesRegex(ValueError, 'Pennies can only be returned for USD fees.'): print(stream.fee.pennies) stream.fee.usd = Decimal('1.01') self.assertEqual(stream.fee.usd, Decimal('1.01')) self.assertEqual(stream.fee.pennies, 101) self.assertEqual(stream.fee.currency, 'USD') stream.fee.pennies = 203 self.assertEqual(stream.fee.usd, Decimal('2.03')) self.assertEqual(stream.fee.pennies, 203) 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) class TestLanguages(TestCase): def test_language_successful_parsing(self): stream = Stream() stream.languages.append('en') self.assertEqual(stream.languages[0].langtag, 'en') self.assertEqual(stream.languages[0].language, 'en') self.assertEqual(stream.langtags, ['en']) stream.languages.append('en-US') self.assertEqual(stream.languages[1].langtag, 'en-US') self.assertEqual(stream.languages[1].language, 'en') self.assertEqual(stream.languages[1].region, 'US') self.assertEqual(stream.langtags, ['en', 'en-US']) stream.languages.append('en-Latn-US') self.assertEqual(stream.languages[2].langtag, 'en-Latn-US') self.assertEqual(stream.languages[2].language, 'en') self.assertEqual(stream.languages[2].script, 'Latn') self.assertEqual(stream.languages[2].region, 'US') self.assertEqual(stream.langtags, ['en', 'en-US', 'en-Latn-US']) stream.languages.append('es-419') self.assertEqual(stream.languages[3].langtag, 'es-419') self.assertEqual(stream.languages[3].language, 'es') self.assertIsNone(stream.languages[3].script) self.assertEqual(stream.languages[3].region, '419') self.assertEqual(stream.langtags, ['en', 'en-US', 'en-Latn-US', 'es-419']) stream = Stream() stream.languages.extend(['en-Latn-US', 'es-ES', 'de-DE']) self.assertEqual(stream.languages[0].language, 'en') self.assertEqual(stream.languages[1].language, 'es') self.assertEqual(stream.languages[2].language, 'de') def test_language_error_parsing(self): stream = Stream() with self.assertRaisesRegex(ValueError, 'Language has no value defined for name zz'): stream.languages.append('zz') with self.assertRaisesRegex(ValueError, 'Script has no value defined for name Zabc'): stream.languages.append('en-Zabc') with self.assertRaisesRegex(ValueError, 'Country has no value defined for name ZZ'): stream.languages.append('en-Zzzz-ZZ') with self.assertRaisesRegex(AssertionError, 'Failed to parse language tag: en-Zzz-US'): stream.languages.append('en-Zzz-US') class TestTags(TestCase): def test_normalize_tags(self): claim = Claim() claim.channel.update(tags=['Anime', 'anime', ' aNiMe', 'maNGA ']) self.assertCountEqual(claim.channel.tags, ['anime', 'manga']) claim.channel.update(tags=['Juri', 'juRi']) self.assertCountEqual(claim.channel.tags, ['anime', 'manga', 'juri']) claim.channel.update(tags='Anime') self.assertCountEqual(claim.channel.tags, ['anime', 'manga', 'juri']) claim.channel.update(clear_tags=True) self.assertEqual(len(claim.channel.tags), 0) claim.channel.update(tags='Anime') self.assertEqual(claim.channel.tags, ['anime']) class TestCollection(TestCase): def test_collection(self): collection = Collection() collection.update(claims=['abc123', 'def123']) self.assertListEqual(collection.claims.ids, ['abc123', 'def123']) collection.update(claims=['abc123', 'bbb123']) self.assertListEqual(collection.claims.ids, ['abc123', 'def123', 'abc123', 'bbb123']) collection.update(clear_claims=True, claims=['bbb987', 'bb']) self.assertListEqual(collection.claims.ids, ['bbb987', 'bb']) self.assertEqual(collection.to_dict(), {'claims': ['bbb987', 'bb']}) collection.update(clear_claims=True) self.assertListEqual(collection.claims.ids, []) class TestLocations(TestCase): def test_location_successful_parsing(self): # from simple string stream = Stream() stream.locations.append('US') self.assertEqual(stream.locations[0].country, 'US') # from full string stream = Stream() stream.locations.append('US:NH:Manchester:03101:42.990605:-71.460989') self.assertEqual(stream.locations[0].country, 'US') self.assertEqual(stream.locations[0].state, 'NH') self.assertEqual(stream.locations[0].city, 'Manchester') self.assertEqual(stream.locations[0].code, '03101') self.assertEqual(stream.locations[0].latitude, '42.990605') self.assertEqual(stream.locations[0].longitude, '-71.460989') # from partial string stream = Stream() stream.locations.append('::Manchester:03101:') self.assertIsNone(stream.locations[0].country) self.assertEqual(stream.locations[0].state, '') self.assertEqual(stream.locations[0].city, 'Manchester') self.assertEqual(stream.locations[0].code, '03101') self.assertIsNone(stream.locations[0].latitude) self.assertIsNone(stream.locations[0].longitude) # from partial string lat/long stream = Stream() stream.locations.append('::::42.990605:-71.460989') self.assertIsNone(stream.locations[0].country) self.assertEqual(stream.locations[0].state, '') self.assertEqual(stream.locations[0].city, '') self.assertEqual(stream.locations[0].code, '') self.assertEqual(stream.locations[0].latitude, '42.990605') self.assertEqual(stream.locations[0].longitude, '-71.460989') # from short circuit lat/long stream = Stream() stream.locations.append('42.990605:-71.460989') self.assertIsNone(stream.locations[0].country) self.assertEqual(stream.locations[0].state, '') self.assertEqual(stream.locations[0].city, '') self.assertEqual(stream.locations[0].code, '') self.assertEqual(stream.locations[0].latitude, '42.990605') self.assertEqual(stream.locations[0].longitude, '-71.460989') # from json string stream = Stream() stream.locations.append('{"country": "ES"}') self.assertEqual(stream.locations[0].country, 'ES') # from dict stream = Stream() stream.locations.append({"country": "UA"}) self.assertEqual(stream.locations[0].country, 'UA') class TestStreamUpdating(TestCase): def test_stream_update(self): stream = Stream() # each of these values is set differently inside of .update() stream.update( title="foo", thumbnail_url="somescheme:some/path", file_name="file-name" ) self.assertEqual(stream.title, "foo") self.assertEqual(stream.thumbnail.url, "somescheme:some/path") self.assertEqual(stream.source.name, "file-name") with self.assertRaises(InputValueIsNoneError): stream.update(title=None) with self.assertRaises(InputValueIsNoneError): stream.update(file_name=None) with self.assertRaises(InputValueIsNoneError): stream.update(thumbnail_url=None)