Tidy and test

This commit is contained in:
Simeon Warner 2020-12-24 13:00:09 -05:00
parent 518fd94e5c
commit 9ffa84a768
3 changed files with 237 additions and 79 deletions

View File

@ -1,11 +1,8 @@
"""ResourceSync Resources - information about a web resource and changes."""
import re
try: # python3
from urllib.parse import urlparse
except ImportError: # python2
from urlparse import urlparse
from posixpath import basename
from urllib.parse import urlparse
from .w3c_datetime import str_to_datetime, datetime_to_str
@ -58,19 +55,21 @@ class Resource(object):
__slots__ = ('uri', 'timestamp', 'length', 'mime_type',
'md5', 'sha1', 'sha256', 'change', 'ts_datetime',
'path', '_extra', 'ln')
'path', 'ln', '_extra')
CHANGE_TYPES = ['created', 'updated', 'deleted']
def __init__(self, uri=None, timestamp=None, length=None,
md5=None, sha1=None, sha256=None, mime_type=None,
change=None, ts_datetime=None, datetime=None, path=None,
lastmod=None, capability=None,
ts_at=None, md_at=None,
ts_completed=None, md_completed=None,
ts_from=None, md_from=None,
ts_until=None, md_until=None,
resource=None, ln=None):
mime_type=None, md5=None, sha1=None, sha256=None,
change=None, ts_datetime=None, path=None, ln=None,
# the following in _extra
capability=None,
ts_at=None, ts_completed=None, ts_from=None, ts_until=None,
# and the following via setters
lastmod=None, datetime=None,
md_at=None, md_completed=None, md_from=None, md_until=None,
# and finally another Resource to copy from
resource=None):
"""Initialize Resource object.
Initialize either from parameters specified or from an existing
@ -88,14 +87,15 @@ class Resource(object):
self.change = None
self.ts_datetime = None # Added in ResourceSync v1.1
self.path = None
self._extra = None
self.ln = None
# Create from a Resource-like object? Copy any relevant attributes
self._extra = None
# Create from a Resource-like object? Copy any attributes, both ones
# that have slots and ones that live in _extra
if (resource is not None):
for att in ['uri', 'timestamp', 'length', 'md5', 'sha1', 'sha256',
'change', 'ts_datetime', 'path', 'capability',
'ts_at', 'md_at', 'ts_completed', 'md_completed',
'ts_from', 'md_from', 'ts_until', 'md_until', 'ln']:
for att in ['uri', 'timestamp', 'length', 'mime_type',
'md5', 'sha1', 'sha256', 'change', 'ts_datetime', 'path', 'ln',
# the following in _extra
'capability', 'ts_at', 'ts_completed', 'ts_from', 'ts_until']:
if hasattr(resource, att):
setattr(self, att, getattr(resource, att))
# Any arguments will then override
@ -146,7 +146,7 @@ class Resource(object):
self.md_until = md_until
# Sanity check
if (self.uri is None):
raise ValueError("Cannot create resource without a URI")
raise ValueError("Cannot create Resource without a URI")
def __setattr__(self, prop, value):
"""Attribute setter with check and support for extra attributes.
@ -199,6 +199,11 @@ class Resource(object):
"""Set timestamp from a W3C Datetime Last-Modified value."""
self.ts_datetime = str_to_datetime(datetime, context='ts_datetime')
@property
def ts_at(self):
"""ts_at is the timestamp for md_at."""
return self._get_extra('ts_at')
@property
def md_at(self):
"""md_at values in W3C Datetime syntax, Z notation."""
@ -211,6 +216,11 @@ class Resource(object):
'ts_at',
str_to_datetime(md_at, context='md_at datetime'))
@property
def ts_completed(self):
"""ts_completed is the timestamp for md_completed."""
return self._get_extra('ts_completed')
@property
def md_completed(self):
"""md_completed value in W3C Datetime syntax, Z notation."""
@ -223,6 +233,11 @@ class Resource(object):
'ts_completed',
str_to_datetime(md_completed, context='md_completed datetime'))
@property
def ts_from(self):
"""ts_from is the timestamp for md_from."""
return self._get_extra('ts_from')
@property
def md_from(self):
"""md_from value in W3C Datetime syntax, Z notation."""
@ -235,6 +250,11 @@ class Resource(object):
'ts_from',
str_to_datetime(md_from, context='md_from datetime'))
@property
def ts_until(self):
"""ts_until is the timestamp for md_until."""
return self._get_extra('ts_until')
@property
def md_until(self):
"""md_until value in W3C Datetime syntax, Z notation."""
@ -277,8 +297,10 @@ class Resource(object):
return(None)
@hash.setter
def hash(self, hash):
"""Parse space separated set of values.
def hash(self, hashes_str):
"""Parse space separated set of values to set md5, sha1 and sha256.
Any existing values for md5, sha1 and sha256 will be removed first.
See specification at:
http://tools.ietf.org/html/draft-snell-atompub-link-extensions-09
@ -287,27 +309,24 @@ class Resource(object):
self.md5 = None
self.sha1 = None
self.sha256 = None
if (hash is None):
return
hash_seen = set()
errors = []
for entry in hash.split():
for entry in hashes_str.split():
(hash_type, value) = entry.split(':', 1)
if (hash_type in hash_seen):
errors.append("Ignored duplicate hash type %s" % (hash_type))
if hash_type in hash_seen:
errors.append("duplicate hash type %s" % (hash_type))
else:
hash_seen.add(hash_type)
if (hash_type == 'md5'):
if hash_type == 'md5':
self.md5 = value
elif (hash_type == 'sha-1'):
elif hash_type == 'sha-1':
self.sha1 = value
elif (hash_type == 'sha-256'):
elif hash_type == 'sha-256':
self.sha256 = value
else:
errors.append("Ignored unsupported hash type (%s)" %
(hash_type))
if (len(errors) > 0):
raise ValueError(". ".join(errors))
errors.append("unsupported hash type %s" % (hash_type))
if len(errors) > 0:
raise ValueError("Ignored " + ", ".join(errors))
def link(self, rel):
"""Look for link with specified rel, return else None.
@ -316,19 +335,16 @@ class Resource(object):
specified rel value. If there are multiple links with the
same rel then just the first will be returned
"""
if (self.ln is None):
return(None)
for link in self.ln:
if ('rel' in link and link['rel'] == rel):
return(link)
return(None)
if self.ln is not None:
for link in self.ln:
if 'rel' in link and link['rel'] == rel:
return link
return None
def link_href(self, rel):
"""Look for link with specified rel, return href from it or None."""
link = self.link(rel)
if (link is not None):
link = link['href']
return(link)
return None if link is None else link['href']
def link_set(self, rel, href, allow_duplicates=False, **atts):
"""Set/create link with specified rel, set href and any other attributes.
@ -342,13 +358,13 @@ class Resource(object):
Be aware that adding links to a Resource object will
significantly increase the size of the object.
"""
if (self.ln is None):
if self.ln is None:
# automagically create a self.ln list
self.ln = []
link = None
else:
link = self.link(rel)
if (link is not None and not allow_duplicates):
if link is not None and not allow_duplicates:
# overwrite current value
link['href'] = href
else:
@ -398,7 +414,7 @@ class Resource(object):
@property
def contents(self):
"""Get the URI of and ResourceSync rel="contents" link."""
return(self.link_href('index'))
return(self.link_href('contents'))
@contents.setter
def contents(self, uri, type='application/xml'):
@ -420,36 +436,43 @@ class Resource(object):
def __eq__(self, other):
"""Equality test for resources allowing <1s difference in timestamp.
See equal(...) for more details of equality test
See equal(...) for more details of equality test.
"""
return(self.equal(other, delta=1.0))
def equal(self, other, delta=0.0):
"""Equality or near equality test for resources.
Equality means:
Equality means that any parameters that exist for both resources match,
essentially "no proof of inequality":
1. same uri, AND
2. same timestamp WITHIN delta if specified for either, AND
3. same md5 if specified for both, AND
3. same md5|sha1|sha256 if specified for both, AND
4. same length if specified for both
"""
if (other is None):
if other is None:
return False
if self.uri != other.uri:
return False
if (self.uri != other.uri):
return(False)
if (self.timestamp is not None or other.timestamp is not None):
# not equal if only one timestamp specified
if (self.timestamp is None
or other.timestamp is None
or abs(self.timestamp - other.timestamp) >= delta):
return(False)
return False
if ((self.md5 is not None and other.md5 is not None)
and self.md5 != other.md5):
return(False)
return False
if ((self.sha1 is not None and other.sha1 is not None)
and self.sha1 != other.sha1):
return False
if ((self.sha256 is not None and other.sha256 is not None)
and self.sha256 != other.sha256):
return False
if ((self.length is not None and other.length is not None)
and self.length != other.length):
return(False)
return(True)
return False
return True
def __str__(self):
"""Return a human readable string for this resource.
@ -458,11 +481,12 @@ class Resource(object):
designed to support logging.
"""
s = [str(self.uri), str(self.lastmod), str(self.length),
str(self.md5 if self.md5 else self.sha1)]
str(self.md5 if self.md5 else (self.sha1 if self.sha1 else self.sha256))]
if (self.change is not None):
s.append(str(self.change))
ch = str(self.change)
if self.datetime is not None:
s.append(" @ " + str(self.datetime))
ch += " @ " + str(self.datetime)
s.append(ch)
if (self.path is not None):
s.append(str(self.path))
return "[ " + " | ".join(s) + " ]"

View File

@ -1,3 +1,4 @@
"""Tests for resync.resource."""
import unittest
import re
from resync.resource import Resource, ChangeTypeError
@ -5,20 +6,60 @@ from resync.resource import Resource, ChangeTypeError
class TestResource(unittest.TestCase):
def test01a_same(self):
def test01_init(self):
"""Test __init__ method."""
# No uri = error
self.assertRaises(ValueError, Resource)
self.assertRaises(ValueError, Resource, timestamp=12)
# Create with many params
r1 = Resource(uri='a', timestamp=0, length=123,
md5='aaa', sha1='bbb', sha256='ccc',
mime_type='text/plain', change='updated',
ts_datetime=1000, path='/a/b/c', ln={'x': 'y'},
ts_at=2000, ts_completed=3000, ts_from=4000,
ts_until=5000)
self.assertEqual(r1.uri, 'a')
self.assertEqual(r1.timestamp, 0)
self.assertEqual(r1.length, 123)
self.assertEqual(r1.md5, 'aaa')
self.assertEqual(r1.sha1, 'bbb')
self.assertEqual(r1.sha256, 'ccc')
self.assertEqual(r1.mime_type, 'text/plain')
self.assertEqual(r1.change, 'updated')
self.assertEqual(r1.ts_datetime, 1000)
self.assertEqual(r1.path, '/a/b/c')
self.assertEqual(r1.ln, {'x': 'y'})
self.assertEqual(r1.ts_at, 2000)
self.assertEqual(r1.ts_completed, 3000)
self.assertEqual(r1.ts_from, 4000)
self.assertEqual(r1.ts_until, 5000)
# Create r2 like r1
r2 = Resource(resource=r1)
self.assertEqual(r2.uri, 'a')
self.assertEqual(r2.timestamp, 0)
self.assertEqual(r2.length, 123)
self.assertEqual(r2.md5, 'aaa')
self.assertEqual(r2.sha1, 'bbb')
self.assertEqual(r2.sha256, 'ccc')
self.assertEqual(r2.mime_type, 'text/plain')
self.assertEqual(r2.change, 'updated')
self.assertEqual(r2.ts_datetime, 1000)
self.assertEqual(r2.path, '/a/b/c')
self.assertEqual(r2.ln, {'x': 'y'})
def test02_equal(self):
"""Test equal method via = operator."""
# just uri
r1 = Resource('a')
r2 = Resource('a')
self.assertEqual(r1, r1)
self.assertEqual(r1, r2)
def test01b_same(self):
# with timestamps
r1 = Resource(uri='a', timestamp=1234.0)
r2 = Resource(uri='a', timestamp=1234.0)
self.assertEqual(r1, r1)
self.assertEqual(r1, r2)
def test01c_same(self):
"""Same with lastmod instead of direct timestamp"""
# with lastmod instead of direct timestamp
r1 = Resource('a')
r1lm = '2012-01-01T00:00:00Z'
r1.lastmod = r1lm
@ -40,26 +81,45 @@ class TestResource(unittest.TestCase):
self.assertEqual(r1.timestamp, r2.timestamp, ('%s (%f) == %s (%f)' % (
r1lm, r1.timestamp, r2lm, r2.timestamp)))
self.assertEqual(r1, r2)
def test01d_same(self):
"""Same with slight timestamp diff"""
# with slight timestamp diff
r1 = Resource('a')
r1.lastmod = '2012-01-02T01:02:03Z'
r2 = Resource('a')
r2.lastmod = '2012-01-02T01:02:03.99Z'
self.assertNotEqual(r1.timestamp, r2.timestamp)
self.assertEqual(r1, r2)
def test02a_diff(self):
# now with too much time diff
r1 = Resource('a', lastmod='2012-01-11')
r2 = Resource('a', lastmod='2012-01-22')
self.assertNotEqual(r1, r2)
# different uris
r1 = Resource('a')
r2 = Resource('b')
self.assertNotEqual(r1, r2)
def test02b_diff(self):
r1 = Resource('a', lastmod='2012-01-11')
r2 = Resource('a', lastmod='2012-01-22')
# print 'r1 == r2 : '+str(r1==r2)
# same and different lengths
r1 = Resource('a', length=1234)
r2 = Resource('a', length=4321)
self.assertNotEqual(r1, r2)
r2.length = r1.md5
self.assertEqual(r1, r2)
# same and different md5
r1.md5 = "3006f84272f2653a6cf5ec3af8f0d773"
r2.md5 = "3006f84272f2653a6cf5ec3af8f00000"
self.assertNotEqual(r1, r2)
r2.md5 = r1.md5
self.assertEqual(r1, r2)
# same and different sha1
r1.sha1 = "3be0f3af2aa4656ce38e0cef305c6eb2af4385d4"
r2.sha1 = "555"
self.assertNotEqual(r1, r2)
r2.sha1 = r1.sha1
self.assertEqual(r1, r2)
# same and different sha256
r1.sha256 = "f41094ad47ef3e93ec1021bfa40f4bf0185f1bf897533638ae5358b61713f84a"
r2.sha256 = "fab"
self.assertNotEqual(r1, r2)
r2.sha256 = r1.sha256
self.assertEqual(r1, r2)
def test04_bad_lastmod(self):
def setlastmod(r, v):
@ -100,7 +160,8 @@ class TestResource(unittest.TestCase):
r1 = Resource('def', lastmod='2012-01-01')
self.assertEqual(repr(r1), "{'uri': 'def', 'timestamp': 1325376000}")
def test08_multiple_hashes(self):
def test08_hash(self):
"""Test hash getter and setters."""
r1 = Resource('abcd')
r1.md5 = "some_md5"
r1.sha1 = "some_sha1"
@ -108,8 +169,7 @@ class TestResource(unittest.TestCase):
self.assertEqual(r1.md5, "some_md5")
self.assertEqual(r1.sha1, "some_sha1")
self.assertEqual(r1.sha256, "some_sha256")
self.assertEqual(
r1.hash, "md5:some_md5 sha-1:some_sha1 sha-256:some_sha256")
self.assertEqual(r1.hash, "md5:some_md5 sha-1:some_sha1 sha-256:some_sha256")
r2 = Resource('def')
r2.hash = "md5:ddd"
self.assertEqual(r2.md5, 'ddd')
@ -121,6 +181,20 @@ class TestResource(unittest.TestCase):
self.assertEqual(r2.md5, 'fff')
self.assertEqual(r2.sha1, 'eee')
self.assertEqual(r2.sha256, 'ggg')
# bogus value will reset
r2.hash = 11
self.assertEqual(r2.md5, None)
self.assertEqual(r2.sha1, None)
self.assertEqual(r2.sha256, None)
# string withough : will raise error
with self.assertRaises(ValueError):
r2.hash = "no-colon"
# dupe
with self.assertRaises(ValueError):
r2.hash = "md5:aaa md5:bbb"
# unknown
with self.assertRaises(ValueError):
r2.hash = "sha999:aaa"
def test09_changetypeerror(self):
r1 = Resource('a')
@ -129,9 +203,11 @@ class TestResource(unittest.TestCase):
self.assertEqual(r1.change, 'deleted')
self.assertRaises(ChangeTypeError, Resource, 'a', change="bad")
# disable checking
ct = Resource.CHANGE_TYPES
Resource.CHANGE_TYPES = False
r1 = Resource('a', change="bad")
self.assertEqual(r1.change, 'bad')
Resource.CHANGE_TYPES = ct
def test10_md_at_roundtrips(self):
r = Resource('a')
@ -182,3 +258,61 @@ class TestResource(unittest.TestCase):
self.assertEqual(r.datetime, None)
r = Resource(uri='dt2', datetime='2000-01-04T00:00:00Z')
self.assertEqual(r.datetime, '2000-01-04T00:00:00Z')
def test15_link(self):
"""Test link link_href and link_set methods."""
r = Resource(uri='ln1')
self.assertEqual(r.link('up'), None)
self.assertEqual(r.link_href('up'), None)
r.link_set('up', 'uri:up')
self.assertEqual(r.link('up'), {'rel': 'up', 'href': 'uri:up'})
self.assertEqual(r.link_href('up'), 'uri:up')
r.link_set('down', 'uri:down')
self.assertEqual(r.link('down'), {'rel': 'down', 'href': 'uri:down'})
self.assertEqual(r.link_href('down'), 'uri:down')
r.link_set('up', 'uri:up2')
self.assertEqual(r.link('up'), {'rel': 'up', 'href': 'uri:up2'})
r.link_add('up', 'uri:up3')
self.assertEqual(r.link('up'), {'rel': 'up', 'href': 'uri:up2'}) # still get first
self.assertEqual(r.ln, [{'rel': 'up', 'href': 'uri:up2'},
{'href': 'uri:down', 'rel': 'down'},
{'rel': 'up', 'href': 'uri:up3'}])
def test16_specific_links(self):
"""Test setters/getters for specific link types."""
r = Resource(uri='laughing')
r.describedby = 'uri:db'
self.assertEqual(r.describedby, 'uri:db')
r.up = 'uri:up'
self.assertEqual(r.up, 'uri:up')
r.index = 'uri:index'
self.assertEqual(r.index, 'uri:index')
r.contents = 'uri:ct'
self.assertEqual(r.contents, 'uri:ct')
def test17_basename(self):
"""Test basename property derived from uri."""
r = Resource(uri='http://example.org/any/complex/path/file')
self.assertEqual(r.basename, 'file')
r.uri = 'http://example.org/any/complex/path/'
self.assertEqual(r.basename, '')
r.uri = 'http://example.org'
self.assertEqual(r.basename, '')
def test18_str(self):
"""Test str method."""
self.assertEqual(str(Resource('uri:a')),
'[ uri:a | None | None | None ]')
self.assertEqual(str(Resource('uri:a', timestamp=0, length=999)),
'[ uri:a | 1970-01-01T00:00:00Z | 999 | None ]')
self.assertEqual(str(Resource('uri:a', timestamp=0, length=999, sha256='abcdef123')),
'[ uri:a | 1970-01-01T00:00:00Z | 999 | abcdef123 ]')
self.assertEqual(str(Resource('uri:a', change='updated', ts_datetime=3661)),
'[ uri:a | None | None | None | updated @ 1970-01-01T01:01:01Z ]')
self.assertEqual(str(Resource('uri:a', path='/a/b/c')),
'[ uri:a | None | None | None | /a/b/c ]')
def test19_change_type_error(self):
"""Test error from bad change type."""
cte = ChangeTypeError('unk')
self.assertIn('ChangeTypeError: got unk, expected one of ', str(cte))

View File

@ -47,7 +47,7 @@ class TestSitemap(unittest.TestCase):
'<url><loc>aardvark</loc><lastmod>2012-01-11T04:05:06Z</lastmod><rs:md datetime="2012-01-11T04:05:06Z" /></url>')
def test_02_resource_str(self):
r1 = Resource('3b', 1234.1, 9999, 'ab54de')
r1 = Resource('3b', 1234.1, length=9999, md5='ab54de')
self.assertEqual(Sitemap().resource_as_xml(r1),
"<url><loc>3b</loc><lastmod>1970-01-01T00:20:34.100000Z</lastmod><rs:md hash=\"md5:ab54de\" length=\"9999\" /></url>")
r1 = Resource('3c', datetime='2013-01-02T13:00:00Z')