swh/model/tests/fields/test_compound.py

@@ -18,12 +18,12 @@ class ValidateCompound(unittest.TestCase):
         def validate_never(model):
             return False
 
-        self.test_model = 'test model'
+        self.test_model = "test model"
         self.test_schema = {
-            'int': (True, simple.validate_int),
-            'str': (True, simple.validate_str),
-            'str2': (True, simple.validate_str),
-            'datetime': (False, simple.validate_datetime),
+            "int": (True, simple.validate_int),
+            "str": (True, simple.validate_str),
+            "str2": (True, simple.validate_str),
+            "datetime": (False, simple.validate_datetime),
             NON_FIELD_ERRORS: validate_always,
         }
 
@@ -31,43 +31,48 @@ class ValidateCompound(unittest.TestCase):
         self.test_schema_shortcut[NON_FIELD_ERRORS] = validate_never
 
         self.test_schema_field_failed = self.test_schema.copy()
-        self.test_schema_field_failed['int'] = (True, [simple.validate_int,
-                                                       validate_never])
+        self.test_schema_field_failed["int"] = (
+            True,
+            [simple.validate_int, validate_never],
+        )
 
         self.test_value = {
-            'str': 'value1',
-            'str2': 'value2',
-            'int': 42,
-            'datetime': datetime.datetime(1990, 1, 1, 12, 0, 0,
-                                          tzinfo=datetime.timezone.utc),
+            "str": "value1",
+            "str2": "value2",
+            "int": 42,
+            "datetime": datetime.datetime(
+                1990, 1, 1, 12, 0, 0, tzinfo=datetime.timezone.utc
+            ),
         }
 
         self.test_value_missing = {
-            'str': 'value1',
+            "str": "value1",
         }
 
         self.test_value_str_error = {
-            'str': 1984,
-            'str2': 'value2',
-            'int': 42,
-            'datetime': datetime.datetime(1990, 1, 1, 12, 0, 0,
-                                          tzinfo=datetime.timezone.utc),
+            "str": 1984,
+            "str2": "value2",
+            "int": 42,
+            "datetime": datetime.datetime(
+                1990, 1, 1, 12, 0, 0, tzinfo=datetime.timezone.utc
+            ),
         }
 
-        self.test_value_missing_keys = {'int'}
+        self.test_value_missing_keys = {"int"}
 
         self.test_value_wrong_type = 42
 
         self.present_keys = set(self.test_value)
-        self.missing_keys = {'missingkey1', 'missingkey2'}
+        self.missing_keys = {"missingkey1", "missingkey2"}
 
     def test_validate_any_key(self):
-        self.assertTrue(
-            compound.validate_any_key(self.test_value, self.present_keys))
+        self.assertTrue(compound.validate_any_key(self.test_value, self.present_keys))
 
         self.assertTrue(
-            compound.validate_any_key(self.test_value,
-                                      self.present_keys | self.missing_keys))
+            compound.validate_any_key(
+                self.test_value, self.present_keys | self.missing_keys
+            )
+        )
 
     def test_validate_any_key_missing(self):
         with self.assertRaises(ValidationError) as cm:
@@ -75,13 +80,13 @@ class ValidateCompound(unittest.TestCase):
 
         exc = cm.exception
         self.assertIsInstance(str(exc), str)
-        self.assertEqual(exc.code, 'missing-alternative-field')
-        self.assertEqual(exc.params['missing_fields'],
-                         ', '.join(sorted(self.missing_keys)))
+        self.assertEqual(exc.code, "missing-alternative-field")
+        self.assertEqual(
+            exc.params["missing_fields"], ", ".join(sorted(self.missing_keys))
+        )
 
     def test_validate_all_keys(self):
-        self.assertTrue(
-            compound.validate_all_keys(self.test_value, self.present_keys))
+        self.assertTrue(compound.validate_all_keys(self.test_value, self.present_keys))
 
     def test_validate_all_keys_missing(self):
         with self.assertRaises(ValidationError) as cm:
@@ -89,41 +94,49 @@ class ValidateCompound(unittest.TestCase):
 
         exc = cm.exception
         self.assertIsInstance(str(exc), str)
-        self.assertEqual(exc.code, 'missing-mandatory-field')
-        self.assertEqual(exc.params['missing_fields'],
-                         ', '.join(sorted(self.missing_keys)))
+        self.assertEqual(exc.code, "missing-mandatory-field")
+        self.assertEqual(
+            exc.params["missing_fields"], ", ".join(sorted(self.missing_keys))
+        )
 
         with self.assertRaises(ValidationError) as cm:
-            compound.validate_all_keys(self.test_value,
-                                       self.present_keys | self.missing_keys)
+            compound.validate_all_keys(
+                self.test_value, self.present_keys | self.missing_keys
+            )
 
         exc = cm.exception
         self.assertIsInstance(str(exc), str)
-        self.assertEqual(exc.code, 'missing-mandatory-field')
-        self.assertEqual(exc.params['missing_fields'],
-                         ', '.join(sorted(self.missing_keys)))
+        self.assertEqual(exc.code, "missing-mandatory-field")
+        self.assertEqual(
+            exc.params["missing_fields"], ", ".join(sorted(self.missing_keys))
+        )
 
     def test_validate_against_schema(self):
         self.assertTrue(
-            compound.validate_against_schema(self.test_model, self.test_schema,
-                                             self.test_value))
+            compound.validate_against_schema(
+                self.test_model, self.test_schema, self.test_value
+            )
+        )
 
     def test_validate_against_schema_wrong_type(self):
         with self.assertRaises(ValidationError) as cm:
-            compound.validate_against_schema(self.test_model, self.test_schema,
-                                             self.test_value_wrong_type)
+            compound.validate_against_schema(
+                self.test_model, self.test_schema, self.test_value_wrong_type
+            )
 
         exc = cm.exception
         self.assertIsInstance(str(exc), str)
-        self.assertEqual(exc.code, 'model-unexpected-type')
-        self.assertEqual(exc.params['model'], self.test_model)
-        self.assertEqual(exc.params['type'],
-                         self.test_value_wrong_type.__class__.__name__)
+        self.assertEqual(exc.code, "model-unexpected-type")
+        self.assertEqual(exc.params["model"], self.test_model)
+        self.assertEqual(
+            exc.params["type"], self.test_value_wrong_type.__class__.__name__
+        )
 
     def test_validate_against_schema_mandatory_keys(self):
         with self.assertRaises(ValidationError) as cm:
-            compound.validate_against_schema(self.test_model, self.test_schema,
-                                             self.test_value_missing)
+            compound.validate_against_schema(
+                self.test_model, self.test_schema, self.test_value_missing
+            )
 
         # The exception should be of the form:
         # ValidationError({
@@ -138,8 +151,8 @@ class ValidateCompound(unittest.TestCase):
             self.assertEqual(len(nested_key), 1)
             nested = nested_key[0]
             self.assertIsInstance(nested, ValidationError)
-            self.assertEqual(nested.code, 'model-field-mandatory')
-            self.assertEqual(nested.params['field'], key)
+            self.assertEqual(nested.code, "model-field-mandatory")
+            self.assertEqual(nested.params["field"], key)
 
     def test_validate_whole_schema_shortcut_previous_error(self):
         with self.assertRaises(ValidationError) as cm:
@@ -176,14 +189,15 @@ class ValidateCompound(unittest.TestCase):
 
         nested = non_field_errors[0]
         self.assertIsInstance(nested, ValidationError)
-        self.assertEqual(nested.code, 'model-validation-failed')
-        self.assertEqual(nested.params['model'], self.test_model)
-        self.assertEqual(nested.params['validator'], 'validate_never')
+        self.assertEqual(nested.code, "model-validation-failed")
+        self.assertEqual(nested.params["model"], self.test_model)
+        self.assertEqual(nested.params["validator"], "validate_never")
 
     def test_validate_against_schema_field_error(self):
         with self.assertRaises(ValidationError) as cm:
-            compound.validate_against_schema(self.test_model, self.test_schema,
-                                             self.test_value_str_error)
+            compound.validate_against_schema(
+                self.test_model, self.test_schema, self.test_value_str_error
+            )
 
         # The exception should be of the form:
         # ValidationError({
@@ -192,21 +206,21 @@ class ValidateCompound(unittest.TestCase):
 
         exc = cm.exception
         self.assertIsInstance(str(exc), str)
-        self.assertEqual(set(exc.error_dict.keys()), {'str'})
+        self.assertEqual(set(exc.error_dict.keys()), {"str"})
 
-        str_errors = exc.error_dict['str']
+        str_errors = exc.error_dict["str"]
         self.assertIsInstance(str_errors, list)
         self.assertEqual(len(str_errors), 1)
 
         nested = str_errors[0]
         self.assertIsInstance(nested, ValidationError)
-        self.assertEqual(nested.code, 'unexpected-type')
+        self.assertEqual(nested.code, "unexpected-type")
 
     def test_validate_against_schema_field_failed(self):
         with self.assertRaises(ValidationError) as cm:
-            compound.validate_against_schema(self.test_model,
-                                             self.test_schema_field_failed,
-                                             self.test_value)
+            compound.validate_against_schema(
+                self.test_model, self.test_schema_field_failed, self.test_value
+            )
 
         # The exception should be of the form:
         # ValidationError({
@@ -215,14 +229,14 @@ class ValidateCompound(unittest.TestCase):
 
         exc = cm.exception
         self.assertIsInstance(str(exc), str)
-        self.assertEqual(set(exc.error_dict.keys()), {'int'})
+        self.assertEqual(set(exc.error_dict.keys()), {"int"})
 
-        int_errors = exc.error_dict['int']
+        int_errors = exc.error_dict["int"]
         self.assertIsInstance(int_errors, list)
         self.assertEqual(len(int_errors), 1)
 
         nested = int_errors[0]
         self.assertIsInstance(nested, ValidationError)
-        self.assertEqual(nested.code, 'field-validation-failed')
-        self.assertEqual(nested.params['validator'], 'validate_never')
-        self.assertEqual(nested.params['field'], 'int')
+        self.assertEqual(nested.code, "field-validation-failed")
+        self.assertEqual(nested.params["validator"], "validate_never")
+        self.assertEqual(nested.params["field"], "int")

swh/model/tests/generate_testdata.py

+# Copyright (C) 2019 The Software Heritage developers
+# See the AUTHORS file at the top-level directory of this distribution
+# License: GNU General Public License version 3, or any later version
+# See top-level LICENSE file for more information
+
+from datetime import datetime
+from random import choice, randint, random, shuffle
+from typing import Dict, List
+
+from pytz import all_timezones, timezone
+
+from swh.model.hashutil import MultiHash
+
+PROTOCOLS = ["git", "http", "https", "deb", "svn", "mock"]
+DOMAINS = ["example.com", "some.long.host.name", "xn--n28h.tld"]
+PATHS = [
+    "",
+    "/",
+    "/stuff",
+    "/stuff/",
+    "/path/to/resource",
+    "/path/with/anchor#id=42",
+    "/path/with/qargs?q=1&b",
+]
+CONTENT_STATUS = ["visible", "hidden", "absent"]
+MAX_DATE = 3e9  # around 2065
+
+
+def gen_all_origins():
+    for protocol in PROTOCOLS:
+        for domain in DOMAINS:
+            for urlpath in PATHS:
+                yield {"url": "%s://%s%s" % (protocol, domain, urlpath)}
+
+
+ORIGINS = list(gen_all_origins())
+
+
+def gen_origins(n: int = 100) -> List:
+    """Returns a list of n randomly generated origins suitable for using as
+    Storage.add_origin() argument.
+
+    """
+    origins = ORIGINS[:]
+    shuffle(origins)
+    return origins[:n]
+
+
+def gen_content():
+    size = randint(1, 10 * 1024)
+    data = bytes(randint(0, 255) for i in range(size))
+    status = choice(CONTENT_STATUS)
+    h = MultiHash.from_data(data)
+    ctime = datetime.fromtimestamp(random() * MAX_DATE, timezone(choice(all_timezones)))
+    content = {
+        "data": data,
+        "status": status,
+        "length": size,
+        "ctime": ctime,
+        **h.digest(),
+    }
+    if status == "absent":
+        content["reason"] = "why not"
+        content["data"] = None
+    return content
+
+
+def gen_contents(n=20) -> List[Dict]:
+    """Returns a list of n randomly generated content objects (as dict) suitable
+    for using as Storage.content_add() argument.
+    """
+    return [gen_content() for i in range(n)]