<commit_before>from __future__ import absolute_import
import unittest
import sys

from io import open
import os
import tempfile
from six import StringIO
from cwltool.main import main
import shutil
import pytest
from cwltool.utils import onWindows

# Module to be tested
from cwltool import provenance

from .util import get_data
import bagit
import posixpath
import ntpath
from six.moves import urllib
from rdflib import Namespace, URIRef, Graph
from rdflib.namespace import RDF,RDFS,SKOS,DCTERMS,FOAF,XSD
import arcp


# RDF namespaces we'll query for later
ORE = Namespace("http://www.openarchives.org/ore/terms/")
PROV = Namespace("http://www.w3.org/ns/prov#")
WFDESC = Namespace("http://purl.org/wf4ever/wfdesc#")
WFPROV = Namespace("http://purl.org/wf4ever/wfprov#")
SCHEMA = Namespace("http://schema.org/")
CWLPROV = Namespace("https://w3id.org/cwl/prov#")




@pytest.mark.skipif(onWindows(),
                    reason="On Windows this would invoke a default docker container")
class TestProvenance(unittest.TestCase):
    folder = None

    def setUp(self):
        self.folder = tempfile.mkdtemp("ro")
        if os.environ.get("DEBUG"):
            print("%s folder: %s" % (this, self.folder))

    def tearDown(self):
        if self.folder and not os.environ.get("DEBUG"):
            #shutil.rmtree(self.folder)
            pass

    def test_hello_workflow(self):
        self.assertEquals(main(['--provenance', self.folder, get_data('tests/wf/hello-workflow.cwl'),
            "--usermessage", "Hello workflow"]), 0)
        self.check_provenance()

    def test_hello_single_tool(self):
        self.assertEquals(main(['--provenance', self.folder, get_data('tests/wf/hello_single_tool.cwl'),
            "--message", "Hello tool"]), 0)
        self.check_provenance(single_tool=True)

    def test_revsort_workflow(self):
        self.assertEquals(main(['--no-container', '--provenance', self.folder, get_data('tests/wf/revsort.cwl'),
            get_data('tests/wf/revsort-job.json')]), 0)
        self.check_provenance()

    def test_nested_workflow(self):
        self.assertEquals(main(['--no-container', '--provenance', self.folder, get_data('tests/wf/nested.cwl')]), 0)
        self.check_provenance(nested=True)

    def check_provenance(self, nested=False, single_tool=False):
        self.check_folders()
        self.check_bagit()
        self.check_ro(nested=nested)
        self.check_prov(nested=nested, single_tool=single_tool)

    def check_folders(self):
        # Our folders
        for d in ("data", "snapshot", "workflow", "metadata", os.path.join("metadata", "provenance")):
            f = os.path.join(self.folder, d)
            self.assertTrue(os.path.isdir(f))

    def check_bagit(self):
        # check bagit structure
        for f in ("bagit.txt", "bag-info.txt", "manifest-sha1.txt", "tagmanifest-sha1.txt", "tagmanifest-sha256.txt"):
            f = os.path.join(self.folder, f)
            self.assertTrue(os.path.isfile(f))
        bag = bagit.Bag(self.folder)
        self.assertTrue(bag.has_oxum())
        (only_manifest, only_fs) = bag.compare_manifests_with_fs()
        self.assertFalse(list(only_manifest), "Some files only in manifest")
        self.assertFalse(list(only_fs), "Some files only on file system")
        missing_tagfiles = bag.missing_optional_tagfiles()
        self.assertFalse(list(missing_tagfiles), "Some files only in tagmanifest")
        bag.validate()
        # TODO: Check other bag-info attributes
        self.assertTrue(arcp.is_arcp_uri(bag.info.get("External-Identifier")))

    def find_arcp(self):
        # First try to find External-Identifier
        bag = bagit.Bag(self.folder)
        ext_id = bag.info.get("External-Identifier")
        if arcp.is_arcp_uri(ext_id):
            return ext_id
        raise Exception("Can't find External-Identifier")
        # return arcp.arcp_random()

    def _arcp2file(self, uri):
        parsed = arcp.parse_arcp(uri)
        # arcp URIs, ensure they are local to our RO
        self.assertEquals(parsed.uuid,
            arcp.parse_arcp(self.find_arcp()).uuid)

        path = parsed.path[1:]  # Strip first /
        # Convert to local path, in case it uses \ on Windows
        lpath = provenance._convert_path(path, posixpath, os.path)
        return os.path.join(self.folder, lpath)

    def check_ro(self, nested=False):
        manifest_file = os.path.join(self.folder, "metadata", "manifest.json")
        self.assertTrue(os.path.isfile(manifest_file), "Can't find " + manifest_file)
        arcp_root = self.find_arcp()
        base = urllib.parse.urljoin(arcp_root, "metadata/manifest.json")
        g = Graph()
        with open(manifest_file, "rb") as f:
            # Note: This will use https://w3id.org/bundle/context
            g.parse(file=f, format="json-ld", publicID=base)
        print("Parsed manifest:\n\n")
        g.serialize(sys.stdout, format="nt")
        ro = None

        for ro in g.subjects(ORE.isDescribedBy, URIRef(base)):
            break
        self.assertTrue(ro, "Can't find RO with ore:isDescribedBy")

        profile = None
        for dc in g.objects(ro, DCTERMS.conformsTo):
            profile = dc
            break
        self.assertTrue(profile, "Can't find profile with dct:conformsTo")
        self.assertEquals(profile, URIRef("https://w3id.org/cwl/prov/0.3.0"),
            "Unexpected cwlprov version " + profile)

        paths = []
        externals = []
        for aggregate in g.objects(ro, ORE.aggregates):
            print(aggregate)
            if not arcp.is_arcp_uri(aggregate):
                externals.append(aggregate)
                # Won't check external URIs existence here
                # TODO: Check they are not relative!
                continue
            lfile = self._arcp2file(aggregate)
            paths.append(os.path.relpath(lfile, self.folder))
            self.assertTrue(os.path.isfile(lfile), "Can't find aggregated " + lfile)

        self.assertTrue(paths, "Didn't find any arcp aggregates")
        self.assertTrue(externals, "Didn't find any data URIs")

        for ext in ["provn", "xml", "json", "jsonld", "nt", "ttl"]:
            f = "metadata/provenance/primary.cwlprov.%s" % ext
            self.assertTrue(f in paths, "provenance file missing " + f)

        for f in ["workflow/primary-job.json", "workflow/packed.cwl"]:
            self.assertTrue(f in paths, "workflow file missing " + f)
        # Can't test snapshot/ files directly as their name varies

        # TODO: check urn:hash::sha1 thingies
        # TODO: Check OA annotations

    def check_prov(self, nested=False, single_tool=False):
        prov_file = os.path.join(self.folder, "metadata", "provenance", "primary.cwlprov.nt")
        self.assertTrue(os.path.isfile(prov_file), "Can't find " + prov_file)
        arcp_root = self.find_arcp()
        # Note: We don't need to include metadata/provnance in base URI
        # as .nt always use absolute URIs
        g = Graph()
        with open(prov_file, "rb") as f:
            g.parse(file=f, format="nt", publicID=arcp_root)
        print("Parsed %s:\n\n" % prov_file)
        g.serialize(sys.stdout, format="nt")
        runs = set(g.subjects(RDF.type, WFPROV.WorkflowRun))

        # master workflow run URI (as urn:uuid:) should correspond to arcp uuid part
        uuid = arcp.parse_arcp(arcp_root).uuid
        master_run = URIRef(uuid.urn)
        self.assertTrue(master_run in runs,
            "Can't find run %s in %s" % (master_run, runs))
        # TODO: we should not need to parse arcp, but follow
        # the has_provenance annotations in manifest.json instead<commit_msg>Test RO "annotations" element<commit_after>from __future__ import absolute_import
import unittest
import sys

from io import open
import os
import tempfile
from six import StringIO
from cwltool.main import main
import shutil
import pytest
from cwltool.utils import onWindows

# Module to be tested
from cwltool import provenance

from .util import get_data
import bagit
import posixpath
import ntpath
from six.moves import urllib
from rdflib import Namespace, URIRef, Graph, Literal
from rdflib.namespace import RDF,RDFS,SKOS,DCTERMS,FOAF,XSD,DC
import arcp


# RDF namespaces we'll query for later
ORE = Namespace("http://www.openarchives.org/ore/terms/")
PROV = Namespace("http://www.w3.org/ns/prov#")
WFDESC = Namespace("http://purl.org/wf4ever/wfdesc#")
WFPROV = Namespace("http://purl.org/wf4ever/wfprov#")
SCHEMA = Namespace("http://schema.org/")
CWLPROV = Namespace("https://w3id.org/cwl/prov#")
OA = Namespace("http://www.w3.org/ns/oa#")



@pytest.mark.skipif(onWindows(),
                    reason="On Windows this would invoke a default docker container")
class TestProvenance(unittest.TestCase):
    folder = None

    def setUp(self):
        self.folder = tempfile.mkdtemp("ro")
        if os.environ.get("DEBUG"):
            print("%s folder: %s" % (this, self.folder))

    def tearDown(self):
        if self.folder and not os.environ.get("DEBUG"):
            #shutil.rmtree(self.folder)
            pass

    def test_hello_workflow(self):
        self.assertEquals(main(['--provenance', self.folder, get_data('tests/wf/hello-workflow.cwl'),
            "--usermessage", "Hello workflow"]), 0)
        self.check_provenance()

    def test_hello_single_tool(self):
        self.assertEquals(main(['--provenance', self.folder, get_data('tests/wf/hello_single_tool.cwl'),
            "--message", "Hello tool"]), 0)
        self.check_provenance(single_tool=True)

    def test_revsort_workflow(self):
        self.assertEquals(main(['--no-container', '--provenance', self.folder, get_data('tests/wf/revsort.cwl'),
            get_data('tests/wf/revsort-job.json')]), 0)
        self.check_provenance()

    def test_nested_workflow(self):
        self.assertEquals(main(['--no-container', '--provenance', self.folder, get_data('tests/wf/nested.cwl')]), 0)
        self.check_provenance(nested=True)

    def check_provenance(self, nested=False, single_tool=False):
        self.check_folders()
        self.check_bagit()
        self.check_ro(nested=nested)
        self.check_prov(nested=nested, single_tool=single_tool)

    def check_folders(self):
        # Our folders
        for d in ("data", "snapshot", "workflow", "metadata", os.path.join("metadata", "provenance")):
            f = os.path.join(self.folder, d)
            self.assertTrue(os.path.isdir(f))

    def check_bagit(self):
        # check bagit structure
        for f in ("bagit.txt", "bag-info.txt", "manifest-sha1.txt", "tagmanifest-sha1.txt", "tagmanifest-sha256.txt"):
            f = os.path.join(self.folder, f)
            self.assertTrue(os.path.isfile(f))
        bag = bagit.Bag(self.folder)
        self.assertTrue(bag.has_oxum())
        (only_manifest, only_fs) = bag.compare_manifests_with_fs()
        self.assertFalse(list(only_manifest), "Some files only in manifest")
        self.assertFalse(list(only_fs), "Some files only on file system")
        missing_tagfiles = bag.missing_optional_tagfiles()
        self.assertFalse(list(missing_tagfiles), "Some files only in tagmanifest")
        bag.validate()
        # TODO: Check other bag-info attributes
        self.assertTrue(arcp.is_arcp_uri(bag.info.get("External-Identifier")))

    def find_arcp(self):
        # First try to find External-Identifier
        bag = bagit.Bag(self.folder)
        ext_id = bag.info.get("External-Identifier")
        if arcp.is_arcp_uri(ext_id):
            return ext_id
        raise Exception("Can't find External-Identifier")
        # return arcp.arcp_random()

    def _arcp2file(self, uri):
        parsed = arcp.parse_arcp(uri)
        # arcp URIs, ensure they are local to our RO
        self.assertEquals(parsed.uuid,
            arcp.parse_arcp(self.find_arcp()).uuid)

        path = parsed.path[1:]  # Strip first /
        # Convert to local path, in case it uses \ on Windows
        lpath = provenance._convert_path(path, posixpath, os.path)
        return os.path.join(self.folder, lpath)

    def check_ro(self, nested=False):
        manifest_file = os.path.join(self.folder, "metadata", "manifest.json")
        self.assertTrue(os.path.isfile(manifest_file), "Can't find " + manifest_file)
        arcp_root = self.find_arcp()
        base = urllib.parse.urljoin(arcp_root, "metadata/manifest.json")
        g = Graph()
        with open(manifest_file, "rb") as f:
            # Note: This will use https://w3id.org/bundle/context
            g.parse(file=f, format="json-ld", publicID=base)
        print("Parsed manifest:\n\n")
        g.serialize(sys.stdout, format="nt")
        ro = None

        for ro in g.subjects(ORE.isDescribedBy, URIRef(base)):
            break
        self.assertTrue(ro, "Can't find RO with ore:isDescribedBy")

        profile = None
        for dc in g.objects(ro, DCTERMS.conformsTo):
            profile = dc
            break
        self.assertTrue(profile, "Can't find profile with dct:conformsTo")
        self.assertEquals(profile, URIRef("https://w3id.org/cwl/prov/0.3.0"),
            "Unexpected cwlprov version " + profile)

        paths = []
        externals = []
        for aggregate in g.objects(ro, ORE.aggregates):
            print(aggregate)
            if not arcp.is_arcp_uri(aggregate):
                externals.append(aggregate)
                # Won't check external URIs existence here
                # TODO: Check they are not relative!
                continue
            lfile = self._arcp2file(aggregate)
            paths.append(os.path.relpath(lfile, self.folder))
            self.assertTrue(os.path.isfile(lfile), "Can't find aggregated " + lfile)

        self.assertTrue(paths, "Didn't find any arcp aggregates")
        self.assertTrue(externals, "Didn't find any data URIs")

        for ext in ["provn", "xml", "json", "jsonld", "nt", "ttl"]:
            f = "metadata/provenance/primary.cwlprov.%s" % ext
            self.assertTrue(f in paths, "provenance file missing " + f)

        for f in ["workflow/primary-job.json", "workflow/packed.cwl"]:
            self.assertTrue(f in paths, "workflow file missing " + f)
        # Can't test snapshot/ files directly as their name varies

        # TODO: check urn:hash::sha1 thingies
        # TODO: Check OA annotations

        packed = urllib.parse.urljoin(arcp_root, "/workflow/packed.cwl")
        primary_job = urllib.parse.urljoin(arcp_root, "/workflow/primary-job.json")
        primary_prov_nt = urllib.parse.urljoin(arcp_root, "/metadata/provenance/primary.cwlprov.nt")
        uuid = arcp.parse_arcp(arcp_root).uuid

        highlights = set(g.subjects(OA.motivatedBy, OA.highlighting))
        self.assertTrue(highlights, "Didn't find highlights")
        for h in highlights:
            self.assertTrue( (h, OA.hasTarget, URIRef(packed)) in g)

        describes = set(g.subjects(OA.motivatedBy, OA.describing))
        for d in describes:
            self.assertTrue( (d, OA.hasBody, URIRef(arcp_root)) in g)
            self.assertTrue( (d, OA.hasTarget, URIRef(uuid.urn)) in g)

        linked = set(g.subjects(OA.motivatedBy, OA.linking))
        for l in linked:
            self.assertTrue( (l, OA.hasBody, URIRef(packed)) in g)
            self.assertTrue( (l, OA.hasBody, URIRef(primary_job)) in g)
            self.assertTrue( (l, OA.hasTarget, URIRef(uuid.urn)) in g)

        has_provenance = set(g.subjects(OA.hasBody, URIRef(primary_prov_nt)))
        for p in has_provenance:
            self.assertTrue( (p, OA.hasTarget, URIRef(uuid.urn)) in g)
            self.assertTrue( (p, OA.motivatedBy, PROV.has_provenance) in g)
            # Check all prov elements are listed
            formats = set()
            for prov in g.objects(p,OA.hasBody):
                self.assertTrue( (prov, DCTERMS.conformsTo,
                                  URIRef("https://w3id.org/cwl/prov/0.3.0") ) in g)
                # NOTE: DC.format is a Namespace method and does not resolve like other terms
                formats.update(set(g.objects(prov, DC["format"])))
            self.assertTrue(formats, "Could not find media types")
            expected = set(Literal(f) for f in (
                "application/json",
                "application/ld+json",
                "application/n-triples",
                'text/provenance-notation; charset="UTF-8"',
                'text/turtle; charset="UTF-8"',
                "application/xml"
            ))
            self.assertEquals(formats, expected,
                "Did not match expected PROV media types")

        if nested:
            # Check for additional PROVs
            # Let's try to find the other wf run ID
            otherRuns = set()
            for p in g.subjects(OA.motivatedBy, PROV.has_provenance):
                if (p, OA.hasTarget, URIRef(uuid.urn)) in g:
                    continue
                otherRuns.update(set(g.objects(p, OA.hasTarget)))
            self.assertTrue(otherRuns, "Could not find nested workflow run prov annotations")

    def check_prov(self, nested=False, single_tool=False):
        prov_file = os.path.join(self.folder, "metadata", "provenance", "primary.cwlprov.nt")
        self.assertTrue(os.path.isfile(prov_file), "Can't find " + prov_file)
        arcp_root = self.find_arcp()
        # Note: We don't need to include metadata/provnance in base URI
        # as .nt always use absolute URIs
        g = Graph()
        with open(prov_file, "rb") as f:
            g.parse(file=f, format="nt", publicID=arcp_root)
        print("Parsed %s:\n\n" % prov_file)
        g.serialize(sys.stdout, format="nt")
        runs = set(g.subjects(RDF.type, WFPROV.WorkflowRun))

        # master workflow run URI (as urn:uuid:) should correspond to arcp uuid part
        uuid = arcp.parse_arcp(arcp_root).uuid
        master_run = URIRef(uuid.urn)
        self.assertTrue(master_run in runs,
            "Can't find run %s in %s" % (master_run, runs))
        # TODO: we should not need to parse arcp, but follow
        # the has_provenance annotations in manifest.json instead
{'fraction_non_alphanumeric': 0.12308032261628549, 'fraction_numerical': 0.0028173682466025854, 'mean_word_length': 2.616983016983017, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 7, 'lorem ipsum': 0, 'www.': 5, 'xml': 3}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '324235', 'n_tokens_mistral': 5862, 'n_tokens_neox': 5695, 'n_words': 1291}