Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found

Target

Select target project
  • anlambert/swh-model
  • lunar/swh-model
  • franckbret/swh-model
  • douardda/swh-model
  • olasd/swh-model
  • swh/devel/swh-model
  • Alphare/swh-model
  • samplet/swh-model
  • marmoute/swh-model
  • rboyer/swh-model
10 results
Show changes
......@@ -16,36 +16,36 @@ def hash_data(raw_content):
class TestValidators(unittest.TestCase):
def setUp(self):
self.valid_visible_content = {
'status': 'visible',
'length': 5,
'data': b'1984\n',
'ctime': datetime.datetime(2015, 11, 22, 16, 33, 56,
tzinfo=datetime.timezone.utc),
"status": "visible",
"length": 5,
"data": b"1984\n",
"ctime": datetime.datetime(
2015, 11, 22, 16, 33, 56, tzinfo=datetime.timezone.utc
),
}
self.valid_visible_content.update(
hash_data(self.valid_visible_content['data']))
self.valid_visible_content.update(hash_data(self.valid_visible_content["data"]))
self.valid_absent_content = {
'status': 'absent',
'length': 5,
'ctime': datetime.datetime(2015, 11, 22, 16, 33, 56,
tzinfo=datetime.timezone.utc),
'reason': 'Content too large',
'sha1_git': self.valid_visible_content['sha1_git'],
'origin': 42,
"status": "absent",
"length": 5,
"ctime": datetime.datetime(
2015, 11, 22, 16, 33, 56, tzinfo=datetime.timezone.utc
),
"reason": "Content too large",
"sha1_git": self.valid_visible_content["sha1_git"],
"origin": 42,
}
self.invalid_content_hash_mismatch = self.valid_visible_content.copy()
self.invalid_content_hash_mismatch.update(
hash_data(b"this is not the data you're looking for"))
hash_data(b"this is not the data you're looking for")
)
def test_validate_content(self):
self.assertTrue(
validators.validate_content(self.valid_visible_content))
self.assertTrue(validators.validate_content(self.valid_visible_content))
self.assertTrue(
validators.validate_content(self.valid_absent_content))
self.assertTrue(validators.validate_content(self.valid_absent_content))
def test_validate_content_hash_mismatch(self):
with self.assertRaises(exceptions.ValidationError) as cm:
......@@ -62,14 +62,17 @@ class TestValidators(unittest.TestCase):
exc = cm.exception
self.assertIsInstance(str(exc), str)
self.assertEqual(set(exc.error_dict.keys()),
{exceptions.NON_FIELD_ERRORS})
self.assertEqual(set(exc.error_dict.keys()), {exceptions.NON_FIELD_ERRORS})
hash_mismatches = exc.error_dict[exceptions.NON_FIELD_ERRORS]
self.assertIsInstance(hash_mismatches, list)
self.assertEqual(len(hash_mismatches), 4)
self.assertTrue(all(mismatch.code == 'content-hash-mismatch'
for mismatch in hash_mismatches))
self.assertEqual(set(mismatch.params['hash']
for mismatch in hash_mismatches),
{'sha1', 'sha1_git', 'sha256', 'blake2s256'})
self.assertTrue(
all(
mismatch.code == "content-hash-mismatch" for mismatch in hash_mismatches
)
)
self.assertEqual(
set(mismatch.params["hash"] for mismatch in hash_mismatches),
{"sha1", "sha1_git", "sha256", "blake2s256"},
)
......@@ -25,8 +25,8 @@ def toposort(revision_log):
# Add the roots to the processing queue.
queue = collections.deque()
for rev in revision_log:
parents = rev['parents']
in_degree[rev['id']] = len(parents)
parents = rev["parents"]
in_degree[rev["id"]] = len(parents)
if not parents:
queue.append(rev)
for parent in parents:
......@@ -37,7 +37,7 @@ def toposort(revision_log):
while queue:
rev = queue.popleft()
yield rev
for child in children[rev['id']]:
in_degree[child['id']] -= 1
if in_degree[child['id']] == 0:
for child in children[rev["id"]]:
in_degree[child["id"]] -= 1
if in_degree[child["id"]] == 0:
queue.append(child)
......@@ -3,8 +3,8 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
from .exceptions import ValidationError, NON_FIELD_ERRORS
from . import fields
from .exceptions import NON_FIELD_ERRORS, ValidationError
from .hashutil import MultiHash, hash_to_bytes
......@@ -14,17 +14,16 @@ def validate_content(content):
Args: a content (dictionary) to validate."""
def validate_content_status(status):
return fields.validate_enum(status, {'absent', 'visible', 'hidden'})
return fields.validate_enum(status, {"absent", "visible", "hidden"})
def validate_keys(content):
hashes = {'sha1', 'sha1_git', 'sha256'}
hashes = {"sha1", "sha1_git", "sha256"}
errors = []
out = True
if content['status'] == 'absent':
if content["status"] == "absent":
try:
out = out and fields.validate_all_keys(content, {'reason',
'origin'})
out = out and fields.validate_all_keys(content, {"reason", "origin"})
except ValidationError as e:
errors.append(e)
try:
......@@ -44,34 +43,36 @@ def validate_content(content):
def validate_hashes(content):
errors = []
if 'data' in content:
hashes = MultiHash.from_data(content['data']).digest()
if "data" in content:
hashes = MultiHash.from_data(content["data"]).digest()
for hash_type, computed_hash in hashes.items():
if hash_type not in content:
continue
content_hash = hash_to_bytes(content[hash_type])
if content_hash != computed_hash:
errors.append(ValidationError(
'hash mismatch in content for hash %(hash)s',
params={'hash': hash_type},
code='content-hash-mismatch',
))
errors.append(
ValidationError(
"hash mismatch in content for hash %(hash)s",
params={"hash": hash_type},
code="content-hash-mismatch",
)
)
if errors:
raise ValidationError(errors)
return True
content_schema = {
'sha1': (False, fields.validate_sha1),
'sha1_git': (False, fields.validate_sha1_git),
'sha256': (False, fields.validate_sha256),
'status': (True, validate_content_status),
'length': (True, fields.validate_int),
'ctime': (True, fields.validate_datetime),
'reason': (False, fields.validate_str),
'origin': (False, fields.validate_int),
'data': (False, fields.validate_bytes),
"sha1": (False, fields.validate_sha1),
"sha1_git": (False, fields.validate_sha1_git),
"sha256": (False, fields.validate_sha256),
"status": (True, validate_content_status),
"length": (True, fields.validate_int),
"ctime": (True, fields.validate_datetime),
"reason": (False, fields.validate_str),
"origin": (False, fields.validate_int),
"data": (False, fields.validate_bytes),
NON_FIELD_ERRORS: [validate_keys, validate_hashes],
}
return fields.validate_against_schema('content', content_schema, content)
return fields.validate_against_schema("content", content_schema, content)
[tox]
envlist=flake8,py3
minversion = 4
envlist =
black
flake8
mypy
py3-{minimal,full-cover}
[testenv]
usedevelop = true
extras =
full: testing
minimal: testing-minimal
deps =
cover: pytest-cov
commands =
pytest --doctest-modules \
cover: --cov={envsitepackagesdir}/swh/model --cov-branch \
full: swh/model \
minimal: swh/model/tests/test_cli.py -m 'not requires_optional_deps' \
{posargs}
[testenv:py3]
[testenv:py3{,7,8,9,10,11,12,13},pypy3{,5,6,7,8,9,10,11,12,13}]
skip_install = true
allowlist_externals = tox
commands =
tox run -e {env_name}-full-cover -- {posargs}
tox run -e {env_name}-minimal
[testenv:black]
skip_install = true
deps =
.[testing]
pytest-cov
black==25.1.0
commands =
pytest --cov=swh --cov-branch {posargs}
{envpython} -m black --check swh
[testenv:flake8]
skip_install = true
deps =
flake8
flake8==7.1.1
flake8-bugbear==24.12.12
flake8-pyproject==1.2.3
pycodestyle==2.12.1
commands =
{envpython} -m flake8
[testenv:mypy]
extras =
testing
deps =
mypy==1.15.0
commands =
mypy swh
# build documentation outside swh-environment using the current
# git HEAD of swh-docs, is executed on CI for each diff to prevent
# breaking doc build
[testenv:sphinx]
allowlist_externals = make
extras =
testing
deps =
# fetch and install swh-docs
git+https://gitlab.softwareheritage.org/swh/devel/swh-docs.git\#egg=swh.docs
setenv =
SWH_PACKAGE_DOC_TOX_BUILD = 1
# turn warnings into errors
SPHINXOPTS = -W
commands =
make -I {env_dir}/share/swh-docs -C docs