rml to html

Author: Dragos0000

ted_sws/rml_to_html/services/rml_to_html.py

@@ -21,12 +21,25 @@ def query_uri_substitution(query: str, triple_map_uri: str) -> str:
 
 
 def get_query_results(query: str, sparql_runner: SPARQLRunner) -> dict:
+    """
+    Method to get query results
+    :param query:
+    :param sparql_runner:
+    :return:
+    """
     return json.loads(
         sparql_runner.query(query_object=query).serialize(
             format="json").decode("utf-8"))["results"]["bindings"]
 
 
 def run_queries_for_triple_map(triple_map_uri: str, query_registry: QueryRegistry, sparql_runner: SPARQLRunner) -> dict:
+    """
+    Running all queries against a triple map URI
+    :param triple_map_uri:
+    :param query_registry:
+    :param sparql_runner:
+    :return:
+    """
     return {
         "triple_map_uri": triple_map_uri,
         "details": get_query_results(
@@ -46,27 +59,27 @@ def run_queries_for_triple_map(triple_map_uri: str, query_registry: QueryRegistr
 
 
 def rml_files_to_html_report(mapping_suite_identifier: str, mapping_suite_repository: MappingSuiteRepositoryABC):
+    """
+    Creating an html report from loaded rml files
+    :param mapping_suite_identifier:
+    :param mapping_suite_repository:
+    :return:
+    """
     mapping_suite_package = mapping_suite_repository.get(reference=mapping_suite_identifier)
     if mapping_suite_package is None:
         raise ValueError(f'Mapping suite package, with {mapping_suite_identifier} id, was not found')
     rml_files = mapping_suite_package.transformation_rule_set.rml_mapping_rules
     query_registry = QueryRegistry()
-
     sparql_runner = SPARQLRunner(files=rml_files)
-    triple_maps = json.loads(sparql_runner.query(query_object=query_registry.TRIPLE_MAP).serialize(
-        format="json").decode("utf-8"))
-    triple_maps_uris = [triple_map['tripleMap']["value"] for triple_map in triple_maps["results"]["bindings"]]
-    print(triple_maps_uris)
+
+    triple_maps = get_query_results(query=query_registry.TRIPLE_MAP, sparql_runner=sparql_runner)
+    triple_maps_uris = [triple_map["tripleMap"]["value"] for triple_map in triple_maps]
     triple_maps_details = {}
     for triple_map_uri in triple_maps_uris:
         triple_maps_details[triple_map_uri] = run_queries_for_triple_map(triple_map_uri=triple_map_uri,
                                                                          query_registry=query_registry,
                                                                          sparql_runner=sparql_runner)
 
     html_report = TEMPLATES.get_template(RML_TO_HTML_REPORT_TEMPLATE).render(triple_maps_details=triple_maps_details)
-    with open("rml-report.html", "w") as file:
-        file.write(html_report)
-
-    return triple_maps_details
-
 
+    return html_report

tests/unit/rml_to_html/conftest.py

@@ -1,8 +1,15 @@
 import pytest
 
+from ted_sws.core.model.transform import FileResource
 from tests import TEST_DATA_PATH
 
 
 @pytest.fixture
 def file_system_repository_path():
-    return TEST_DATA_PATH / "notice_transformer" / "test_repository"
+    return TEST_DATA_PATH / "notice_transformer" / "test_repository"
+
+
+@pytest.fixture
+def rml_file():
+    path = TEST_DATA_PATH / "notice_transformer" / "test_repository" / "test_package" / "transformation" / "mappings" / "award_of_contract.rml.ttl"
+    return FileResource(file_name="file_name", file_content=path.read_text())

tests/unit/rml_to_html/test_rml_files_to_html.py

@@ -1,14 +1,38 @@
-import json
+import pytest
 
 from ted_sws.data_manager.adapters.mapping_suite_repository import MappingSuiteRepositoryInFileSystem
-from ted_sws.rml_to_html.services.rml_to_html import rml_files_to_html_report
+from ted_sws.notice_validator.adapters.sparql_runner import SPARQLRunner
+from ted_sws.rml_to_html.resources.query_registry import QueryRegistry
+from ted_sws.rml_to_html.services.rml_to_html import rml_files_to_html_report, query_uri_substitution, get_query_results
+
+
+def test_query_uri_substitution():
+    query = QueryRegistry().SUBJECT_MAP
+    uri = "http://data.europa.eu/a4g/mapping/sf-rml/ResultNotice"
+    substituted_query = query_uri_substitution(query=query,
+                                               triple_map_uri=uri)
+    assert uri in substituted_query
+
+
+def test_get_query_results(rml_file):
+    query = QueryRegistry().TRIPLE_MAP
+    results = get_query_results(query=query, sparql_runner=SPARQLRunner(files=[rml_file]))
+    assert isinstance(results, list)
+    for result in results:
+        assert isinstance(result, dict)
+        assert "type", "value" in result.keys()
 
 
 def test_rml_files_to_html_report(file_system_repository_path):
     mapping_suite_repository = MappingSuiteRepositoryInFileSystem(repository_path=file_system_repository_path)
-    triple_maps=rml_files_to_html_report(mapping_suite_identifier="test_package",mapping_suite_repository=mapping_suite_repository)
+    html_report = rml_files_to_html_report(mapping_suite_identifier="test_package",
+                                           mapping_suite_repository=mapping_suite_repository)
+
+    assert isinstance(html_report, str)
+    assert "Logical Source" in html_report
+    assert "Subject map" in html_report
+    assert "Predicate object maps" in html_report
 
-    # triple_maps_uris = [triple_map['tripleMap']["value"] for triple_map in triple_maps["results"]["bindings"]]
-    # #
-    # with open("triple-maps.json", "w") as file:
-    #     file.write(json.dumps(triple_maps))
+    with pytest.raises(ValueError):
+        rml_files_to_html_report(mapping_suite_identifier="no_package",
+                                 mapping_suite_repository=mapping_suite_repository)