semra/tests/test_neo4j_output.py at ef318540d811d7388b8e015ef51af068dcde28bb · biopragmatics/semra · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
"""Test Neo4j output."""

import tempfile
import unittest
from pathlib import Path

import semra
from semra import (
    EXACT_MATCH,
    LEXICAL_MAPPING,
    MANUAL_MAPPING,
    UNSPECIFIED_MAPPING,
    Mapping,
    MappingSet,
    ReasonedEvidence,
    Reference,
    SimpleEvidence,
)
from semra.io import write_neo4j
from semra.struct import Triple
from semra.vocabulary import BEN_REFERENCE, CHAIN_MAPPING, CHARLIE
from tests import resources

# TODO test when concept name has problematic characters like tabs or newlines


class TestNeo4jOutput(unittest.TestCase):
    """Test Neo4j output."""

    def test_neo4j_output(self) -> None:
        """Test Neo4j output."""
        r1 = Reference.from_curie("mesh:C406527", name="R 115866")
        r2 = Reference.from_curie("chebi:101854", name="talarozole")
        r3 = Reference.from_curie("chembl.compound:CHEMBL459505", name="TALAROZOLE")

        t1 = Triple(subject=r1, predicate=EXACT_MATCH, object=r2)
        t2 = Triple(subject=r2, predicate=EXACT_MATCH, object=r3)
        t3 = Triple(subject=r1, predicate=EXACT_MATCH, object=r3)

        biomappings_purl = "https://w3id.org/biopragmatics/biomappings/sssom/biomappings.sssom.tsv"
        biomappings = MappingSet(
            purl=biomappings_purl,
            name="biomappings",
            confidence=0.90,
            license="CC0",
        )
        # self.assertEqual("cb30f2dd2a144ad3cf877d14e9e88dbf", biomappings.hexdigest())

        chembl = MappingSet(
            name="chembl",
            confidence=0.90,
            license="CC-BY-SA-3.0",
        )

        m1_e1 = SimpleEvidence(
            mapping_set=biomappings,
            justification=MANUAL_MAPPING,
            author=CHARLIE,
            confidence=0.99,
        )
        # self.assertEqual("b59546c8b03b27da7e89b6a08c76843b", m1_e1.hexdigest(t1))

        # check that making an identical evidence gives the same hex digest
        m1_e1_copy = SimpleEvidence(
            mapping_set=biomappings,
            justification=MANUAL_MAPPING,
            author=CHARLIE,
            confidence=0.99,
        )
        self.assertEqual(m1_e1.hexdigest(t1), m1_e1_copy.hexdigest(t1))

        m1_e2 = SimpleEvidence(
            mapping_set=biomappings,
            justification=MANUAL_MAPPING,
            author=BEN_REFERENCE,
            confidence=0.94,
        )
        m1_e3 = SimpleEvidence(
            mapping_set=MappingSet(
                name="lexical",
                confidence=0.90,
            ),
            justification=LEXICAL_MAPPING,
            confidence=0.8,
        )

        m1 = Mapping.from_triple(
            t1,
            evidence=[m1_e1, m1_e2, m1_e3],
        )

        # this curie is generated as a md5 digest of the pickle dump
        # of the 3-tuple of CURIE strings for the subject, predicate, object
        # m1_hexdigest = "9f85f585b0179ba53df2dd274e0067dc"
        # self.assertEqual(m1_hexdigest, m1.hexdigest())

        expected_hex = Mapping.from_triple(t1).hexdigest()

        # Test that the evidences don't affect the hash
        for x in [
            Mapping.from_triple(t1, evidence=[m1_e1]),
            Mapping.from_triple(t1, evidence=[m1_e2]),
            Mapping.from_triple(t1, evidence=[m1_e3]),
            Mapping.from_triple(t1, evidence=[m1_e2, m1_e1]),
            Mapping.from_triple(t1, evidence=[m1_e2, m1_e1, m1_e3]),
        ]:
            self.assertEqual(expected_hex, x.hexdigest())

        m2_e1 = SimpleEvidence(
            mapping_set=chembl,
            justification=UNSPECIFIED_MAPPING,
            confidence=0.90,
        )
        m2 = Mapping.from_triple(t2, evidence=[m2_e1])

        m3_e1 = ReasonedEvidence(justification=CHAIN_MAPPING, mappings=[m1, m2])
        m3_e1_rev = ReasonedEvidence(justification=CHAIN_MAPPING, mappings=[m2, m1])
        m3 = Mapping.from_triple(t3, evidence=[m3_e1])
        # m3_hexdigest = "6185a77934184112529b68ea7780a54d"
        # self.assertEqual(m3_hexdigest, m3.hexdigest())

        # check that order of mappings in evidence doesn't change the hash
        self.assertEqual(
            Mapping.from_triple(t3, evidence=[m3_e1]).hexdigest(),
            Mapping.from_triple(t3, evidence=[m3_e1_rev]).hexdigest(),
        )

        mappings: list[semra.Mapping] = [m1, m2, m3]

        with tempfile.TemporaryDirectory() as _directory:
            directory = Path(_directory)

            write_neo4j(mappings, directory, use_tqdm=False)
            # FIXME this test is important since it makes sure we get deterministic hashes each time
            #  but changes in the underlying data structure updated the pickles, invaliding old
            #  hashes. Therefore, we should stop relying on picking for hashing and define explicit
            #  ones for every object (yes, obvious in hindsight)
            for path in [
                resources.CONCEPT_NODES_TSV_PATH,
                resources.EVIDENCE_NODES_TSV_PATH,
                resources.MAPPING_NODES_TSV_PATH,
                resources.MAPPING_SET_NODES_TSV_PATH,
                resources.EDGES_TSV_PATH,
                resources.MAPPING_EDGES_TSV_PATH,
            ]:
                self.assertEqual(path.read_text(), directory.joinpath(path.name).read_text())