Refs #9 -- Clean up RPSL parser script somewhat and added a test.

.coveragerc

@@ -0,0 +1 @@
+setup.cfg

irrd/rpsl/rpsl_objects.py

@@ -15,7 +15,7 @@ def rpsl_object_from_text(text, strict_validation=True):
     try:
         klass = OBJECT_CLASS_MAPPING[rpsl_object_class]
     except KeyError:
-        raise UnknownRPSLObjectClassException(f"Encountered unknown object class {rpsl_object_class}")
+        raise UnknownRPSLObjectClassException(f"Encountered unknown object class: {rpsl_object_class}")
     return klass(from_text=text, strict_validation=strict_validation)
 
 

irrd/scripts/rpsl_parse.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python
+"""
+This is a small helper script to run RPSL data through the parser.
+
+This may be useful to validate the strictness of the parser against
+live RPSL data.
+"""
+import argparse
+import sys
+
+from irrd.rpsl.parser import UnknownRPSLObjectClassException
+from irrd.rpsl.rpsl_objects import rpsl_object_from_text
+
+
+obj_parsed = 0
+obj_errors = 0
+obj_unknown = 0
+unknown_object_classes = set()
+
+
+def parse(rpsl_text, strict_validation):
+    global obj_parsed
+    global obj_errors
+    global obj_unknown
+
+    if not rpsl_text.strip():
+        return
+    try:
+        obj_parsed += 1
+        obj = rpsl_object_from_text(rpsl_text.strip(), strict_validation=strict_validation)
+        if obj.messages.messages():
+            if obj.messages.errors():
+                obj_errors += 1
+            print(rpsl_text.strip())
+            print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
+            print(obj.messages)
+            print("\n=======================================\n")
+    except UnknownRPSLObjectClassException as e:
+        obj_unknown += 1
+        unknown_object_classes.add(str(e).split(":")[1].strip())
+    except Exception as e:  # pragma: no cover
+        print("=======================================")
+        print(input)
+        print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~")
+        raise e
+
+
+def main(filename, strict_validation):
+    global obj_parsed
+    global obj_errors
+    global obj_unknown
+    global unknown_object_classes
+
+    obj_parsed = 0
+    obj_errors = 0
+    obj_unknown = 0
+    unknown_object_classes = set()
+
+    if filename == '-':  # pragma: no cover
+        f = sys.stdin
+    else:
+        f = open(filename, encoding="iso-8859-1")
+
+    current_obj = ""
+    for line in f.readlines():
+        if line.startswith("%") or line.startswith("#"):
+            continue
+        current_obj += line
+
+        if not line.strip("\r\n"):
+            parse(current_obj, strict_validation)
+            current_obj = ""
+
+    parse(current_obj, strict_validation)
+
+    print(f"Processed {obj_parsed} objects, {obj_errors} with errors")
+    if obj_unknown:
+        unknown_formatted = ', '.join(unknown_object_classes)
+        print(f"Ignored {obj_unknown} objects due to unknown object classes: {unknown_formatted}")
+
+
+if __name__ == "__main__":
+    description = """Run RPSL data through the IRRD parser. For each object that resulted in messages emitted by
+                     the parser, the object is printed followed by the messages."""
+    parser = argparse.ArgumentParser(description=description)
+    parser.add_argument("input_file", type=str,
+                        help="the name of a file to read, or - for stdin")
+    parser.add_argument("--strict", dest="strict_validation", action="store_true",
+                        help="use strict validation (errors on e.g. unknown or missing attributes)")
+    args = parser.parse_args()
+
+    main(args.input_file, args.strict_validation)

irrd/scripts/tests/test_rpsl_parse.py

@@ -0,0 +1,60 @@
+from ..rpsl_parse import main
+
+# First an entirely valid object with an as-block that should be reformatted,
+# then an object with an extra unkown attributes,
+# then an entirely unknown object.
+TEST_DATA = """# TEST
+% TEST
+
+as-block:       AS2043 - as02043
+descr:          RIPE NCC ASN block
+remarks:        These AS Numbers are assigned to network operators in the RIPE NCC service region.
+mnt-by:         RIPE-NCC-HM-MNT
+changed:        2014-02-24T13:15:13Z
+tech-c:         DUMY-RIPE
+admin-c:        DUMY-RIPE
+source:         RIPE
+
+as-block:       AS2043 - as02043
+descr:          RIPE NCC ASN block
+remarks:        These AS Numbers are assigned to network operators in the RIPE NCC service region.
+mnt-by:         RIPE-NCC-HM-MNT
+changed:        2014-02-24T13:15:13Z
+tech-c:         DUMY-RIPE
+unknown-obj:    unknown value should be caught in strict validation
+admin-c:        DUMY-RIPE
+source:         RIPE
+
+foo-block:       AS2043 - as02043
+descr:          RIPE NCC ASN block
+remarks:        These AS Numbers are assigned to network operators in the RIPE NCC service region.
+mnt-by:         RIPE-NCC-HM-MNT
+changed:        2014-02-24T13:15:13Z
+tech-c:         DUMY-RIPE
+admin-c:        DUMY-RIPE
+source:         RIPE
+
+
+"""
+
+
+def test_rpsl_parse(capsys, tmpdir):
+    tmp_file = tmpdir + "/rpsl_parse_test.rpsl"
+    fh = open(tmp_file, "w")
+    fh.write(TEST_DATA)
+    fh.close()
+
+    main(filename=tmp_file, strict_validation=True)
+    captured = capsys.readouterr().out
+    assert "ERROR: Unrecognised attribute unknown-obj on object as-block" in captured
+    assert "INFO: AS range AS2043 - as02043 was reformatted as AS2043 - AS2043" in captured
+    assert "Processed 3 objects, 1 with errors" in captured
+    assert "Ignored 1 objects due to unknown object classes: foo-block" in captured
+
+    main(filename=tmp_file, strict_validation=False)
+    captured = capsys.readouterr().out
+    print(captured)
+    assert "ERROR: Unrecognised attribute unknown-obj on object as-block" not in captured
+    assert "INFO: AS range AS2043 - as02043 was reformatted as AS2043 - AS2043" in captured
+    assert "Processed 3 objects, 0 with errors" in captured
+    assert "Ignored 1 objects due to unknown object classes: foo-block" in captured

setup.cfg

@@ -1,2 +1,13 @@
 [flake8]
 ignore=E501,W503,E226
+
+[report]
+exclude_lines =
+    pragma: no cover
+    def __repr__
+    if self.debug:
+    if settings.DEBUG
+    raise AssertionError
+    raise NotImplementedError
+    if 0:
+    if __name__ == "__main__":