prodcom-data/tests/test_prodcom_correspondence.py

#!/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: <http://w3id.org/probs-lab/data/prodcom/>
                 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: <http://w3id.org/probs-lab/data/prodcom/>
                 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: <http://w3id.org/probs-lab/data/prodcom/>
                 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