#!/usr/bin/env python3
import pytest
from numpy import isnan
from hashlib import sha256
from rdflib import Namespace
from probs_runner import PROBS, QUANTITYKIND, answer_queries
from numpy.testing import assert_allclose
from decimal import Decimal
#from uuid import uuid4
PRODCOM2016 = Namespace("http://w3id.org/probs-lab/data/prodcom/2016/")
PRODCOM2017 = Namespace("http://w3id.org/probs-lab/data/prodcom/2017/")
GEONAMES = Namespace("https://sws.geonames.org/")
def assert_exact_results(result, expected_value):
assert len(result) == 1
assert result[0].bound == PROBS.ExactBound
assert_allclose(result[0].measurement, expected_value, rtol=1e-3)
def assert_no_measurement(result):
assert len(result) == 1
assert isnan(result[0].measurement)
class TestProdcomCorrespondence:
"""Test PRODCOM Correspondence File Conversion"""
endpoint_data = ["outputs/PRC_2017_2016.nt.gz",
"outputs/PRD_2016_20200617_185122.nt.gz",
"outputs/PRD_2017_20200617_185035.nt.gz"
]
@pytest.mark.parametrize("code", [
("10112000"), ("22197120"), ("26601170")
])
def test_object_equivalence(self, rdfox, code):
# object has same code in PRODCOM 2016 and 2017 lists
obj_name = "\"PRODCOM Object from Code " + code + "\""
query = r""" PREFIX ufpc:
SELECT ?a ?b WHERE {
?a rdfs:label %s ;
:hasClassificationCode ?c.
?c :partOfList ufpc:Prodcom2016List .
?b rdfs:label %s ;
:hasClassificationCode ?d.
?d :partOfList ufpc:Prodcom2017List .
?b :objectEquivalentTo ?a .
} """ % (obj_name, obj_name)
result = rdfox.query_records(query)
assert len(result) == 1
@pytest.mark.parametrize("code", [
("20201100"), ("265100Z0"), ("28491290")
])
def test_new_object(self, rdfox, code):
# new object in PRODCOM 2017 list
obj_name = "\"PRODCOM Object from Code " + code + "\""
query = r""" PREFIX ufpc:
SELECT ?a WHERE {
?a rdfs:label %s ;
:hasClassificationCode ?c .
?c :partOfList ufpc:Prodcom2017List .
FILTER NOT EXISTS {
?b :hasClassificationCode ?d .
?d :partOfList ufpc:Prodcom2016List .
?a :objectEquivalentTo ?b .
}
} """ % (obj_name)
result = rdfox.query_records(query)
assert len(result) == 1
@pytest.mark.parametrize("code", [
("23311073"), ("28302370"), ("32403990")
])
def test_object_deleted(self, rdfox, code):
# object in PRODCOM 2016 list deleted in 2017 list
obj_name = "\"PRODCOM Object from Code " + code + "\""
query = r""" PREFIX ufpc:
SELECT ?a WHERE {
?a rdfs:label %s ;
:hasClassificationCode ?c .
?c :partOfList ufpc:Prodcom2016List .
FILTER NOT EXISTS {
?b :hasClassificationCode ?d .
?d :partOfList ufpc:Prodcom2017List .
?b :objectEquivalentTo ?a .
}
} """ % (obj_name)
result = rdfox.query_records(query)
assert len(result) == 1