Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ with open("out.ged", "wb") as f:

Each record is a `GedcomStructure` with a `tag`, an optional `xref` and `pointer`, the raw `text` payload, and `children`. Its `value` property casts the payload to the data type the specification gives that structure type.

`loads` and `dumps` are the string equivalents. Non-conforming input raises `GedcomParseError`, a `ValueError` carrying `line_number`.
`loads` and `dumps` are the string equivalents. Non-conforming input raises `GedcomParseError`, a `ValueError` carrying `line_number`. A pointer to a record that does not exist is kept rather than rejected; `gedcom7.validate(records)` lists each one as a `dangling-pointer` error.

## Development

Expand Down
48 changes: 25 additions & 23 deletions gedcom7/parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,15 @@
from __future__ import annotations

import re
import sys
from typing import TYPE_CHECKING

from . import const, grammar
from .exceptions import GedcomParseError
from .types import GedcomStructure

if TYPE_CHECKING:
from collections.abc import Iterator
from typing import BinaryIO

# EOL = %x0D [%x0A] / %x0A -- CR-LF, CR, or LF
Expand All @@ -32,6 +34,21 @@ def _unescape(linestr: str) -> str:
return linestr[1:] if linestr.startswith("@@") else linestr


def _lines(string: str) -> Iterator[str]:
"""Yield the lines of a data stream, without their EOLs, one at a time.

A terminating EOL does not start a further, empty line. A data stream whose
last line lacks its EOL is tolerated: the line is complete and unambiguous,
and dropping it would silently lose data.
"""
start = 0
for eol in _EOL.finditer(string):
yield string[start : eol.start()]
start = eol.end()
if start < len(string):
yield string[start:]


def load(fp: BinaryIO) -> list[GedcomStructure]:
"""Load a GEDCOM 7 dataset from a binary file object.

Expand All @@ -56,6 +73,10 @@ def loads(string: str) -> list[GedcomStructure]:

Raises :class:`~gedcom7.exceptions.GedcomParseError` if the data stream does
not conform to the specification. Non-conforming lines are never skipped.

A pointer that matches no cross-reference identifier is kept as written, so
one broken pointer does not cost the rest of the dataset;
:func:`~gedcom7.validate` reports each one as a ``dangling-pointer`` error.
Comment thread
DavidMStraub marked this conversation as resolved.
"""
string = string.removeprefix(_BOM)

Expand All @@ -66,32 +87,26 @@ def loads(string: str) -> list[GedcomStructure]:
line_number=string.count("\n", 0, banned.start()) + 1,
)

lines = _EOL.split(string)
# A terminating EOL leaves a final empty element that is not itself a line.
# A data stream whose last line lacks its EOL is tolerated: the line is
# complete and unambiguous, and dropping it would silently lose data.
if lines and lines[-1] == "":
lines.pop()

records: list[GedcomStructure] = []
# stack[i] is the structure encoded by the nearest preceding line of level i
stack: list[GedcomStructure] = []
# extension tag -> URIs declared for it by the header schema
schema: dict[str, list[str]] = {}
xrefs: set[str] = set()
pointers: list[tuple[str, int]] = []
# the structure a CONT on the very next line would continue
continuable: GedcomStructure | None = None

for number, text in enumerate(lines, start=1):
for number, text in enumerate(_lines(string), start=1):
match = _LINE.fullmatch(text + "\n")
if match is None:
raise GedcomParseError(
f"malformed line: {text!r}", line_number=number, line=text
)

level = int(match.group("level"))
tag = match.group("tag")
# A dataset repeats a few dozen tags across every line, so each line's
# tag shares one string object rather than holding its own copy.
tag = sys.intern(match.group("tag"))
xref = match.group("xref")
pointer = match.group("pointer")
linestr = match.group("linestr")
Expand Down Expand Up @@ -188,19 +203,6 @@ def loads(string: str) -> list[GedcomStructure]:
stack.append(structure)
continuable = structure

if pointer is not None and pointer != const.VOIDPTR:
pointers.append((pointer, number))

# Pointers may be forward references, so they are resolved once the whole
# data stream has been read.
for pointer, number in pointers:
if pointer not in xrefs:
raise GedcomParseError(
f"pointer {pointer} matches no cross-reference identifier in the "
"data stream",
line_number=number,
)

if not records:
raise GedcomParseError("a dataset must contain a header and a trailer")
if records[0].tag != const.HEAD:
Expand Down
2 changes: 1 addition & 1 deletion gedcom7/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
from . import cast, const


@dataclass
@dataclass(slots=True)
class GedcomStructure:
"""Gedcom structure class."""

Expand Down
32 changes: 18 additions & 14 deletions gedcom7/validator.py
Original file line number Diff line number Diff line change
Expand Up @@ -201,17 +201,18 @@ def _check_structure(
structure,
)
_check_substructure(structure, errors)
if type_id is not None:
payload = const.payloads.get(type_id)
if payload is not None:
_check_payload_kind(structure, payload, errors)
# Only one of these can say anything useful. Where the payload is of
# the wrong kind entirely, chasing its pointer or casting its text
# would report the same mistake a second time.
if payload.startswith("@<"):
_check_pointer(structure, payload, xrefs, errors)
else:
_check_payload_value(structure, type_id, errors)
payload = const.payloads.get(type_id) if type_id is not None else None
points = payload is not None and payload.startswith("@<")
if type_id is not None and payload is not None:
_check_payload_kind(structure, payload, errors)
# A pointer payload's text is not cast: where it holds the pointer by
# mistake, the kind check above has already said so.
if not points:
_check_payload_value(structure, type_id, errors)
# Any structure may carry a pointer, whether the specification gives it a
# pointer payload, a value, or nothing it knows of, as for an extension. A
# pointer to a missing record is reported wherever it is found.
_check_pointer(structure, payload if points else None, xrefs, errors)
for child in structure.children:
_check_structure(child, xrefs, declared, errors)

Expand Down Expand Up @@ -267,11 +268,14 @@ def _check_payload_kind(

def _check_pointer(
structure: types.GedcomStructure,
payload: str,
payload: str | None,
xrefs: dict[str, types.GedcomStructure],
errors: list[Error],
) -> None:
"""Check that a pointer reaches a record, and one of the right type."""
"""Check that a pointer reaches a record, and one of the type ``payload`` names.

With no ``payload``, as for an extension, any record will do.
"""
if not structure.pointer or structure.pointer == const.VOIDPTR:
return
target = xrefs.get(structure.pointer)
Expand All @@ -282,7 +286,7 @@ def _check_pointer(
f"{structure.pointer} is not a record in this dataset",
structure,
)
else:
elif payload is not None:
wanted = payload[2:-2]
if target.type_id != wanted:
_report(
Expand Down
25 changes: 20 additions & 5 deletions test/test_conformance.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
"""Conformance tests against the FamilySearch GEDCOM 7 specification.

Each test cites the requirement it covers. Cases that the specification
prohibits must raise, not be silently skipped or reinterpreted.
prohibits must raise, not be silently skipped or reinterpreted. The exception
is a pointer to a missing record, which parses and is reported by ``validate``.
"""

import pytest
Expand Down Expand Up @@ -185,10 +186,24 @@ def test_pointer_and_voidptr() -> None:
assert indi.children[1].pointer == "@VOID@"


def test_unresolved_pointer_rejected() -> None:
"""A pointer must match an xref of a structure within the document."""
with pytest.raises(GedcomParseError, match="matches no cross-reference"):
parse("0 @I1@ INDI\n1 ALIA @I9@\n")
def test_unresolved_pointers_parse_and_validate_reports_each() -> None:
"""A pointer must match an xref of a structure within the document.

The parser keeps an unresolved pointer as written, and ``validate`` reports
every one, not only the first.
"""
records = parse(
"0 @I1@ INDI\n1 ALIA @I9@\n1 FAMS @F9@\n1 FAMC @VOID@\n"
"0 @F1@ FAM\n1 HUSB @I1@\n1 WIFE @I8@\n"
)
assert records[1].children[0].pointer == "@I9@"
assert records[1].children[1].pointer == "@F9@"
errors = [e for e in gedcom7.validate(records) if e.category == "dangling-pointer"]
assert [e.path for e in errors] == [
"@I1@ INDI > ALIA",
"@I1@ INDI > FAMS",
"@F1@ FAM > WIFE",
]


def test_forward_pointer_allowed() -> None:
Expand Down
15 changes: 15 additions & 0 deletions test/test_parser.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
import itertools
import pathlib
import re

import gedcom7
from gedcom7.parser import _lines

GEDCOM_MIN = """0 HEAD
1 GEDC
Expand Down Expand Up @@ -103,3 +106,15 @@ def test_empty_line_value() -> None:
assert file_struct.tag == "FILE"
assert file_struct.text == ""
assert file_struct.children[0].tag == "FORM"


def test_lines_matches_splitting_on_every_eol() -> None:
"""Every string over a character, CR and LF, up to seven long, splits into
the lines that splitting on EOL and dropping a final empty piece gives."""
for n in range(8):
for chars in itertools.product("a\r\n", repeat=n):
string = "".join(chars)
expected = re.split(r"\r\n|\r|\n", string)
if expected[-1] == "":
expected.pop()
assert list(_lines(string)) == expected, repr(string)
67 changes: 66 additions & 1 deletion test/test_validator.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Tests for reporting what is wrong with a dataset."""

import pathlib
from collections.abc import Callable

import pytest

Expand Down Expand Up @@ -62,6 +63,70 @@ def test_void_pointer_is_not_dangling() -> None:
assert gedcom7.validate(records) == []


def _pointer_standard(pointer: str) -> types.GedcomStructure:
return individual(types.GedcomStructure(tag="FAMS", pointer=pointer))


def _pointer_on_a_value(pointer: str) -> types.GedcomStructure:
return individual(types.GedcomStructure(tag="SEX", pointer=pointer))


def _pointer_on_no_payload(pointer: str) -> types.GedcomStructure:
return individual(types.GedcomStructure(tag="BAPL", pointer=pointer))


def _pointer_documented_extension(pointer: str) -> types.GedcomStructure:
return individual(types.GedcomStructure(tag=FOAF, pointer=pointer))


def _pointer_undocumented_extension(pointer: str) -> types.GedcomStructure:
return individual(types.GedcomStructure(tag="_MINE", pointer=pointer))


def _pointer_below_an_extension(pointer: str) -> types.GedcomStructure:
extension = types.GedcomStructure(tag="_MINE")
extension.append_child(types.GedcomStructure(tag="FAMS", pointer=pointer))
return individual(extension)


def _pointer_misplaced_standard_tag(pointer: str) -> types.GedcomStructure:
birth = types.GedcomStructure(tag="BIRT")
birth.append_child(types.GedcomStructure(tag="FAMS", pointer=pointer))
return individual(birth)


def _pointer_on_a_record(pointer: str) -> types.GedcomStructure:
return types.GedcomStructure(tag="_REC", xref="@X1@", pointer=pointer)


# Every place a pointer can sit, by what the tables know of its structure: a
# pointer payload, a value payload, no payload, a documented or undocumented
# extension, a standard tag with no standard type here, or a record.
POINTER_HOLDERS = [
_pointer_standard,
_pointer_on_a_value,
_pointer_on_no_payload,
_pointer_documented_extension,
_pointer_undocumented_extension,
_pointer_below_an_extension,
_pointer_misplaced_standard_tag,
_pointer_on_a_record,
]


@pytest.mark.parametrize("holder", POINTER_HOLDERS)
@pytest.mark.parametrize(
("pointer", "dangles"), [("@F9@", True), ("@VOID@", False), ("@I1@", False)]
)
def test_dangling_pointer_is_reported_wherever_it_sits(
holder: Callable[[str], types.GedcomStructure], pointer: str, dangles: bool
) -> None:
"""loads keeps a pointer to a missing record, so validate must find every one."""
built = holder(pointer)
records = dataset(built) if built.tag == "INDI" else dataset(individual(), built)
assert ("dangling-pointer" in categories(records)) is dangles


def test_pointer_at_the_wrong_record_type() -> None:
"""FAMS names a family, so pointing it at an individual is wrong."""
other = types.GedcomStructure(tag="INDI", xref="@I2@")
Expand Down Expand Up @@ -125,7 +190,7 @@ def test_text_where_a_pointer_belongs() -> None:


def test_pointer_where_text_belongs() -> None:
records = dataset(individual(types.GedcomStructure(tag="SEX", pointer="@I2@")))
records = dataset(individual(types.GedcomStructure(tag="SEX", pointer="@I1@")))
assert categories(records) == ["misplaced-payload"]


Expand Down
Loading