diff --git a/README.md b/README.md
index 27835f8..1aec699 100644
--- a/README.md
+++ b/README.md
@@ -68,6 +68,8 @@ second_site:
password1: password1
user2: user2
password2: password2
+ solrurl: http://someserver:8080/solr
+ triplestoreurl: http://someserver:8080/fuseki/test/sparql
```
To use the `second_site` configuration, simply start the testrunner with
@@ -82,11 +84,13 @@ You can also choose to run only a subset of all tests using the `-t|--tests` arg
of the following values which indicate which tests to run.
* `authz` - Authorization tests
* `basic` - Basic interaction tests
-* `sparql` - Sparql operation tests
+* `camel` - Camel toolbox tests (see [note](#camel-tests))
+* `fixity` - Binary fixity tests
+* `indirect` - Indirect container tests
* `rdf` - RDF serialization tests
-* `version` - Versioning tests
+* `sparql` - Sparql operation tests
* `transaction` - Transcation tests
-* `fixity` - Binary fixity tests
+* `version` - Versioning tests
Without this parameter all the above tests will be run.
@@ -95,6 +99,14 @@ To run only the `authz` and `sparql` tests you would execute:
./testrunner.py -c config.yml -t authz,sparql
```
+##### Camel Tests
+`camel` tests are **NOT** executed by default, due to timing issues they should be run separately.
+
+They also require the configuration to have a `solrurl` parameter pointing to a Solr endpoint and a
+`triplestoreurl` parameter pointing to the SPARQL endpoint of a triplestore.
+
+Both of these systems must be fed by the fcrepo-camel-toolbox for this testing.
+
## Tests implemented
### authz
@@ -108,6 +120,34 @@ To run only the `authz` and `sparql` tests you would execute:
1. Verify regular user 1 can access **cover**
1. Verify regular user 2 can't access **cover**
+
+1. Create a container that is readonly for regular user 1
+1. Create a container that regular user 1 has read/write access to.
+1. Verify that regular user 1 can create/edit/append the container
+1. Verify that regular user 1 cannot create a direct or indirect container
+that targets the read-only container as the membership resource.
+
+
+1. Create a container
+1. Add an acl with multiple authorizations for user 1
+1. Verify that user 1 receives the most permissive set of permissions
+from the authorizations
+
+
+1. Verify that the `rel="acl"` link header is the same for:
+ * a binary
+ * its description
+ * the binary timemap
+ * the description timemap
+ * a binary memento
+ * a description memento
+
+
+1. Verify that both a binary and its description share the permissions give
+to the binary.
+
+
+
### basic
1. Create a container
1. Create a container inside the container from step 1
@@ -121,11 +161,39 @@ To run only the `authz` and `sparql` tests you would execute:
1. Create a LDP Indirect container
1. Validate the correct Link header type
+
+1. Create a basic container
+1. Create an indirect container
+1. Create a direct container
+1. Create a NonRDFSource
+1. Try to create a ldp:Resource (not allowed)
+1. Try to create a ldp:Container (not allowed)
+
+
+1. Try to change each of the following (basic, direct, indirect container
+and binary) to all other types.
+
+### camel - see [note](#camel-tests)
+1. Create a container
+1. Check the container is indexed to Solr
+1. Check the container is indexed to the triplestore
+
### fixity
1. Create a binary resource
1. Get a fixity result for that resource
1. Compare that the SHA-1 hash matches the expected value
+### indirect
+1. Create a pcdm:Object
+2. Create a pcdm:Collection
+3. Create an indirect container "members" inside the pcdm:Collection
+4. Create a proxy object for the pcdm:Object inside the **members** indirectContainer
+5. Verify that the pcdm:Collection has the memberRelation property added pointing to the pcdm:Object
+
+### rdf
+1. Create a RDFSource object.
+1. Retrieve that object in all possible RDF serializations.
+
### sparql
1. Create a container
1. Set the dc:title of the container with a Patch request
@@ -137,7 +205,7 @@ To run only the `authz` and `sparql` tests you would execute:
1. Verify the title
1. Create a container
1. Update the title to text with Unicode characters
-1. Verify the title
+1. Verify the title
### transaction
1. Create a transaction
@@ -170,9 +238,3 @@ To run only the `authz` and `sparql` tests you would execute:
1. Create Memento at deleted memento's datetime
1. Verify Memento exists
-### indirect
-1. Create a pcdm:Object
-2. Create a pcdm:Collection
-3. Create an indirect container "members" inside the pcdm:Collection
-4. Create a proxy object for the pcdm:Object inside the **members** indirectContainer
-5. Verify that the pcdm:Collection has the memberRelation property added pointing to the pcdm:Object
diff --git a/TestConstants.py b/TestConstants.py
index 7cee1b8..382456b 100644
--- a/TestConstants.py
+++ b/TestConstants.py
@@ -10,15 +10,18 @@
USER2_PASS_PARAM = "password2"
LOG_FILE_PARAM = "logfile"
SELECTED_TESTS_PARAM = "selected_tests"
+SOLR_URL_PARAM = "solrurl"
+TRIPLESTORE_URL_PARAM = "triplestoreurl"
# Via RFC 7231 3.3
PAYLOAD_HEADERS = ['Content-Length', 'Content-Range', 'Trailer', 'Transfer-Encoding']
# Fedora specific constants
+FEDORA_NS = "http://fedora.info/definitions/v4/repository#"
FCR_VERSIONS = "fcr:versions"
FCR_FIXITY = "fcr:fixity"
-SERVER_MANAGED = "http://fedora.info/definitions/v4/repository#ServerManaged"
-INBOUND_REFERENCE = "http://fedora.info/definitions/v4/repository#InboundReferences"
-EMBEDED_RESOURCE = "http://fedora.info/definitions/v4/repository#EmbedResources"
+SERVER_MANAGED = FEDORA_NS + "ServerManaged"
+INBOUND_REFERENCE = FEDORA_NS + "InboundReferences"
+EMBEDED_RESOURCE = FEDORA_NS + "EmbedResources"
GET_PREFER_MINIMAL = "return=minimal"
PUT_PREFER_LENIENT = "handling=lenient; received=\"minimal\""
@@ -31,6 +34,10 @@
# General Mime and LDP constants
JSONLD_MIMETYPE = "application/ld+json"
SPARQL_UPDATE_MIMETYPE = "application/sparql-update"
+TURTLE_MIMETYPE = "text/turtle"
+SPARQL_QUERY_MIMETYPE = "application/sparql-query"
+SPARQL_RESULT_JSON_MIMETYPE = "application/sparql-results+json"
+LINK_FORMAT_MIMETYPE = "application/link-format"
LDP_NS = "http://www.w3.org/ns/ldp#"
LDP_CONTAINER = LDP_NS + "Container"
@@ -44,9 +51,21 @@
MEM_ORIGINAL_RESOURCE = MEMENTO_NS + "OriginalResource"
MEM_TIMEGATE = MEMENTO_NS + "TimeGate"
MEM_TIMEMAP = MEMENTO_NS + "TimeMap"
+MEM_MEMENTO = MEMENTO_NS + "Memento"
+
+ACL_NS = "http://www.w3.org/ns/auth/acl#"
+
+PURL_NS = "http://purl.org/dc/elements/1.1/"
# Test constructs
OBJECT_TTL = "@prefix dc: ." \
"@prefix pcdm: ." \
"<> a pcdm:Object ;" \
"dc:title \"An Object\" ."
+
+PCDM_CONTAINER_TITLE = "PCDM Container"
+
+PCDM_CONTAINER_TTL = "@prefix dc: ." \
+ "@prefix pcdm: ." \
+ "<> a pcdm:Object ;" \
+ "dc:title \"{0}\" .".format(PCDM_CONTAINER_TITLE)
diff --git a/abstract_fedora_tests.py b/abstract_fedora_tests.py
index 1e500c2..2d1f697 100644
--- a/abstract_fedora_tests.py
+++ b/abstract_fedora_tests.py
@@ -33,14 +33,23 @@ def getFedoraBase(self):
""" Return the Fedora Base URI """
return self.config[TestConstants.BASE_URL_PARAM]
+ @staticmethod
+ def getCurrentClass():
+ return inspect.stack()[1][0].f_locals['self'].__class__.__name__
+
def run_tests(self):
+ current = FedoraTests.getCurrentClass()
self.check_for_retest(self.getBaseUri())
+ self.log("\nStarting class {0}\n".format(current))
""" Check we can access the machine and then run the tests """
self.not_authorized()
for test in self._testdict:
method = getattr(self, test)
+ self.log("Running {0}".format(test))
method()
+ self.log("Passed\n")
self.cleanup(self.getBaseUri())
+ self.log("\nExiting class {0}".format(current))
def not_authorized(self):
""" Ensure we can even access the repository """
@@ -71,6 +80,11 @@ def create_user_auth(self):
return FedoraTests.create_auth(
self.config[TestConstants.USER_NAME_PARAM], self.config[TestConstants.USER_PASS_PARAM])
+ def create_user2_auth(self):
+ """ Create a Basic Auth object using the test user 2 username:password """
+ return FedoraTests.create_auth(
+ self.config[TestConstants.USER2_NAME_PARAM], self.config[TestConstants.USER2_PASS_PARAM])
+
def get_auth(self, admin=True):
""" The admin argument of this function is used through out the testing infrastructure, it is explained here.
True - use admin username / password credentials
@@ -138,6 +152,7 @@ def do_options(self, url, admin=True):
return requests.options(url, auth=my_auth)
def assert_regex_in(self, pattern, container, msg):
+ """ Do a regex match against all members of a list """
for i in container:
if re.search(pattern, i):
return
@@ -146,13 +161,15 @@ def assert_regex_in(self, pattern, container, msg):
self.fail(self._formatMessage(msg, standard_msg))
def assert_regex_matches(self, pattern, text, msg):
+ """ Do a regex match against a string """
if re.search(pattern, text):
return
standard_msg = '%s pattern not matched in %s' % (unittest.util.safe_repr(pattern),
unittest.util.safe_repr(text))
self.fail(self._formatMessage(msg, standard_msg))
- def make_type(self, type):
+ @staticmethod
+ def make_type(type):
""" Turn a URI to Link type format """
return "<{0}>; rel=\"type\"".format(type)
@@ -181,21 +198,29 @@ def cleanup(self, uri):
def check_for_retest(self, uri):
"""Try to create CONTAINER """
- response = self.do_put(uri)
- if response.status_code != 201:
- caller = inspect.stack()[1][0].f_locals['self'].__class__.__name__
- print("The class ({}) has been run.\nYou need to remove the resource ({}) and all it's "
- "children before re-running the test.".format(caller, uri))
- rerun = input("Remove the test objects and re-run? (y/N) ")
- if rerun.lower().strip() == 'y':
- if self.cleanup(uri):
- self.do_put(uri)
+ try:
+ response = self.do_put(uri)
+ if response.status_code == 403:
+ print("Received a 403 Forbidden response, please check your credentials and try again.")
+ quit()
+ elif response.status_code != 201:
+ caller = FedoraTests.getCurrentClass()
+ print("The class ({0}) has been run.\nYou need to remove the resource ({1}) and all it's "
+ "children before re-running the test.".format(caller, uri))
+ rerun = input("Remove the test objects and re-run? (y/N) ")
+ if rerun.lower().strip() == 'y':
+ if self.cleanup(uri):
+ self.do_put(uri)
+ else:
+ print("Error removing {0}, you may need to remove it manually.".format(uri))
+ quit()
else:
- print("Error removing $URL, you may need to remove it manually.")
+ print("Exiting...")
quit()
- else:
- print("Exiting...")
- quit()
+ except requests.exceptions.ConnectionError:
+ self.log("Unable to connect to your repository, if you are sure its running. Please check your base uri "
+ "in the configuration.")
+ quit()
@staticmethod
def get_link_headers(response):
@@ -262,7 +287,8 @@ def assertTitleExists(self, expected, location):
return
self.fail("Did not find expected title \"{0}\" in response".format(expected))
- def log(self, message):
+ @staticmethod
+ def log(message):
print(message)
def find_binary_description(self, response):
@@ -270,6 +296,13 @@ def find_binary_description(self, response):
self.assertIsNotNone(headers['describedby'])
return headers['describedby'][0]
+ def checkResponse(self, expected, response):
+ self.checkValue(expected, response.status_code)
+
+ def checkValue(self, expected, received):
+ self.assertEqual(expected, received, "Did not get expected value")
+ FedoraTests.log(" Passed {0} == {0}".format(received))
+
def Test(func):
""" Decorator for isolating test functions """
diff --git a/authz_tests.py b/authz_tests.py
index acf40ca..d0ada43 100644
--- a/authz_tests.py
+++ b/authz_tests.py
@@ -34,24 +34,27 @@ def verifyAuthEnabled(self):
temp_auth = FedoraTests.create_auth(random_string, random_string)
r = self.do_get(self.getFedoraBase(), admin=temp_auth)
- self.assertEqual(401, r.status_code, "Did not get expected response code")
-
- def getAclUri(self, response):
- acls = self.get_link_headers(response)
- self.assertIsNotNone(acls['acl'])
- return acls['acl'][0]
+ if 401 != r.status_code:
+ self.log("It appears that authentication is not enabled on your repository.")
+ quit()
+
+ @staticmethod
+ def getAclUri(response):
+ acls = FedoraAuthzTests.get_link_headers(response)
+ try:
+ return acls['acl'][0]
+ except KeyError:
+ Exception("No acl link header found")
@Test
def doAuthTests(self):
- self.log("Running doAuthTests")
-
self.verifyAuthEnabled()
self.log("Create \"cover\" container")
r = self.do_put(self.getBaseUri() + "/cover")
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
cover_location = self.get_location(r)
- cover_acl = self.getAclUri(r)
+ cover_acl = FedoraAuthzTests.getAclUri(r)
self.log("Make \"cover\" a pcdm:Object")
sparql = "PREFIX pcdm: " \
@@ -61,11 +64,11 @@ def doAuthTests(self):
'Content-type': TestConstants.SPARQL_UPDATE_MIMETYPE
}
r = self.do_patch(cover_location, headers=headers, body=sparql)
- self.assertEqual(204, r.status_code, "Did not get expected response code")
+ self.checkResponse(204, r)
self.log("Verify no current ACL")
r = self.do_get(cover_acl)
- self.assertEqual(404, r.status_code, "Did not get expected response code")
+ self.checkResponse(404, r)
self.log("Add ACL to \"cover\"")
headers = {
@@ -73,51 +76,51 @@ def doAuthTests(self):
}
body = self.COVER_ACL.format(cover_location, self.config[TestConstants.USER_NAME_PARAM])
r = self.do_put(cover_acl, headers=headers, body=body)
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
self.log("Create \"files\" inside \"cover\"")
r = self.do_put(cover_location + "/files")
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
files_location = self.get_location(r)
- files_acl = self.getAclUri(r)
+ files_acl = FedoraAuthzTests.getAclUri(r)
self.log("Anonymous can't access \"cover\"")
r = self.do_get(cover_location, admin=None)
- self.assertEqual(401, r.status_code, "Did not get expected response code")
+ self.checkResponse(401, r)
self.log("Anonymous can't access \"cover/files\"")
r = self.do_get(files_location, admin=None)
- self.assertEqual(401, r.status_code, "Did not get expected response code")
+ self.checkResponse(401, r)
self.log("{0} can access \"cover\"".format(self.config[TestConstants.ADMIN_USER_PARAM]))
r = self.do_get(cover_location)
- self.assertEqual(200, r.status_code, "Did not get expected response code")
+ self.checkResponse(200, r)
self.log("{0} can access \"cover/files\"".format(self.config[TestConstants.ADMIN_USER_PARAM]))
r = self.do_get(files_location)
- self.assertEqual(200, r.status_code, "Did not get expected response code")
+ self.checkResponse(200, r)
self.log("{0} can access \"cover\"".format(self.config[TestConstants.USER_NAME_PARAM]))
r = self.do_get(cover_location, admin=False)
- self.assertEqual(200, r.status_code, "Did not get expected response code")
+ self.checkResponse(200, r)
self.log("{0} can access \"cover/files\"".format(self.config[TestConstants.USER_NAME_PARAM]))
r = self.do_get(files_location, admin=False)
- self.assertEqual(200, r.status_code, "Did not get expected response code")
+ self.checkResponse(200, r)
auth = self.create_auth(self.config[TestConstants.USER2_NAME_PARAM],
self.config[TestConstants.USER2_PASS_PARAM])
self.log("{0} can't access \"cover\"".format(self.config[TestConstants.USER2_NAME_PARAM]))
r = self.do_get(cover_location, admin=auth)
- self.assertEqual(403, r.status_code, "Did not get expected response code")
+ self.checkResponse(403, r)
self.log("{0} can't access \"cover/files\"".format(self.config[TestConstants.USER2_NAME_PARAM]))
r = self.do_get(files_location, admin=auth)
- self.assertEqual(403, r.status_code, "Did not get expected response code")
+ self.checkResponse(403, r)
self.log("Verify \"cover/files\" has no ACL")
r = self.do_get(files_acl)
- self.assertEqual(404, r.status_code, "Did not get expected response code")
+ self.checkResponse(404, r)
self.log("PUT Acl to \"cover/files\" to allow access for {0}".format(self.config[TestConstants.USER2_NAME_PARAM]))
headers = {
@@ -125,102 +128,100 @@ def doAuthTests(self):
}
body = self.FILES_ACL.format(files_location, self.config[TestConstants.USER2_NAME_PARAM])
r = self.do_put(files_acl, headers=headers, body=body)
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
self.log("{0} can't access \"cover\"".format(self.config[TestConstants.USER2_NAME_PARAM]))
r = self.do_get(cover_location, admin=auth)
- self.assertEqual(403, r.status_code, "Did not get expected response code")
+ self.checkResponse(403, r)
self.log("{0} can access \"cover/files\"".format(self.config[TestConstants.USER2_NAME_PARAM]))
r = self.do_get(files_location, admin=auth)
- self.assertEqual(200, r.status_code, "Did not get expected response code")
-
- self.log("Passed")
+ self.checkResponse(200, r)
@Test
def doDirectIndirectAuthTests(self):
- self.log("Running doDirectIndirectAuthTests")
-
self.verifyAuthEnabled()
self.log("Create a target container")
r = self.do_post()
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
target_location = self.get_location(r)
- target_acl = self.getAclUri(r)
+ target_acl = FedoraAuthzTests.getAclUri(r)
self.log("Create a write container")
r = self.do_post()
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
write_location = self.get_location(r)
- write_acl = self.getAclUri(r)
+ write_acl = FedoraAuthzTests.getAclUri(r)
self.log("Make sure the /target resource is readonly")
- target_ttl = "@prefix acl: .\n"\
+ target_ttl = "@prefix acl: <{2}> .\n"\
"<#readauthz> a acl:Authorization ;\n" \
" acl:agent \"{0}\" ;\n" \
" acl:mode acl:Read ;\n" \
- " acl:accessTo <{1}> .\n".format(self.config[TestConstants.USER_NAME_PARAM], target_location)
+ " acl:accessTo <{1}> .\n".format(self.config[TestConstants.USER_NAME_PARAM], target_location,
+ TestConstants.ACL_NS)
headers = {
'Content-type': 'text/turtle'
}
r = self.do_put(target_acl, headers=headers, body=target_ttl)
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
self.log("Make sure the write resource is writable by \"{0}\"".format(self.config[TestConstants.USER_NAME_PARAM]))
- write_ttl = "@prefix acl: .\n" \
+ write_ttl = "@prefix acl: <{2}> .\n" \
"<#writeauth> a acl:Authorization ;\n" \
" acl:agent \"{0}\" ;\n" \
" acl:mode acl:Read, acl:Write ;\n" \
" acl:accessTo <{1}> ;\n" \
- " acl:default <{1}> .\n".format(self.config[TestConstants.USER_NAME_PARAM], write_location)
+ " acl:default <{1}> .\n".format(self.config[TestConstants.USER_NAME_PARAM], write_location,
+ TestConstants.ACL_NS)
r = self.do_put(write_acl, headers=headers, body=write_ttl)
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
self.log("Verify that \"{0}\" can create a simple resource under write resource (POST)".format(self.config[TestConstants.USER_NAME_PARAM]))
r = self.do_post(write_location, admin=False)
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
uuid_value = str(uuid.uuid4())
self.log("Verify that \"{0}\" can create a simple resource under write resource (PUT)".format(
self.config[TestConstants.USER_NAME_PARAM]))
r = self.do_put(write_location + "/" + uuid_value, admin=False)
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
self.log("Verify that \"{0}\" CANNOT create a resource under target resource".format(self.config[TestConstants.USER_NAME_PARAM]))
r = self.do_post(target_location, admin=False)
- self.assertEqual(403, r.status_code, "Did not get expected response code")
+ self.checkResponse(403, r)
self.log("Verify that \"{0}\" CANNOT create direct or indirect containers that reference target resources".format(self.config[TestConstants.USER_NAME_PARAM]))
headers = {
'Content-type': 'text/turtle',
'Link': self.make_type(TestConstants.LDP_DIRECT)
}
- direct_ttl = "@prefix ldp: .\n" \
+ direct_ttl = "@prefix ldp: <{0}> .\n" \
"@prefix test: .\n" \
- "<> ldp:membershipResource <{0}> ;\n" \
- "ldp:hasMemberRelation test:predicateToCreate .\n".format(target_location)
+ "<> ldp:membershipResource <{1}> ;\n" \
+ "ldp:hasMemberRelation test:predicateToCreate .\n".format(TestConstants.LDP_NS, target_location)
r = self.do_post(write_location, headers=headers, body=direct_ttl, admin=False)
- self.assertEqual(403, r.status_code, "Did not get expected response code")
+ self.checkResponse(403, r)
headers = {
'Content-type': 'text/turtle',
'Link': self.make_type(TestConstants.LDP_INDIRECT)
}
- indirect_ttl = "@prefix ldp: .\n" \
+ indirect_ttl = "@prefix ldp: <{0}> .\n" \
"@prefix test: .\n" \
"<> ldp:insertedContentRelation test:something ;\n" \
- "ldp:membershipResource <{0}> ;\n" \
- "ldp:hasMemberRelation test:predicateToCreate .\n".format(target_location)
+ "ldp:membershipResource <{1}> ;\n" \
+ "ldp:hasMemberRelation test:predicateToCreate .\n".format(TestConstants.LDP_NS, target_location)
r = self.do_post(write_location, headers=headers, body=indirect_ttl, admin=False)
- self.assertEqual(403, r.status_code, "Did not get expected response code")
+ self.checkResponse(403, r)
self.log("Go ahead and create the indirect and direct containers as admin")
r = self.do_post(write_location, headers=headers, body=direct_ttl)
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
direct_location = self.get_location(r)
r = self.do_post(write_location, headers=headers, body=indirect_ttl)
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
indirect_location = self.get_location(r)
self.log("Attempt to verify that \"{0}\" can not actually create relationships on the readonly resource via " \
@@ -232,10 +233,240 @@ def doDirectIndirectAuthTests(self):
self.log("Verify that \"{0}\" can still create a simple resource under write resource (POST)".format(self.config[TestConstants.USER_NAME_PARAM]))
r = self.do_post(write_location, admin=False)
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
uuid_value = str(uuid.uuid4())
self.log("Verify that \"{0}\" can still create a simple resource under write resource (PUT)".format(
self.config[TestConstants.USER_NAME_PARAM]))
r = self.do_put(write_location + "/" + uuid_value, admin=False)
- self.assertEqual(201, r.status_code, "Did not get expected response code")
+ self.checkResponse(201, r)
+
+ @Test
+ def multipleAuthzCreatePermissiveSet(self):
+ self.verifyAuthEnabled()
+
+ self.log("Create a target container")
+ r = self.do_post()
+ self.checkResponse(201, r)
+ target_location = self.get_location(r)
+ target_acl = FedoraAuthzTests.getAclUri(r)
+
+ double_ttl = "@prefix acl: <{2}> .\n" \
+ "<#readonly> a acl:Authorization ;\n" \
+ " acl:agent \"{0}\" ;\n" \
+ " acl:mode acl:Read ;\n" \
+ " acl:accessTo <{1}> .\n" \
+ "<#readwrite> a acl:Authorization ;\n" \
+ " acl:agent \"{0}\" ;\n" \
+ " acl:mode acl:Write ;\n" \
+ " acl:accessTo <{1}> .\n".format(self.config[TestConstants.USER_NAME_PARAM], target_location,
+ TestConstants.ACL_NS)
+ headers = {
+ 'Content-type': TestConstants.TURTLE_MIMETYPE
+ }
+
+ self.log("Add ACL with one read and one write authz for the same URI.")
+ r = self.do_put(target_acl, headers=headers, body=double_ttl)
+ self.checkResponse(201, r)
+
+ self.log("Check we can read")
+ r = self.do_get(target_location, admin=False)
+ self.checkResponse(200, r)
+
+ self.log("Check we can write")
+ headers = {
+ 'Content-type': TestConstants.SPARQL_UPDATE_MIMETYPE
+ }
+ body = "prefix dc: <{0}> INSERT {{ <> dc:title \"A new title\" }} WHERE {{}}".format(TestConstants.PURL_NS)
+ r = self.do_patch(target_location, headers=headers, body=body)
+ self.checkResponse(204, r)
+
+ @Test
+ def testAllThingsPointTogether(self):
+ self.verifyAuthEnabled()
+
+ self.log("Create a target binary")
+ headers = {
+ 'Content-type': 'text/plain',
+ 'Link': self.make_type(TestConstants.LDP_NON_RDF_SOURCE)
+ }
+ r = self.do_post(headers=headers, body="this is a test payload")
+ self.checkResponse(201, r)
+ binary_location = self.get_location(r)
+ expected_acl = binary_location + "/fcr:acl"
+ acl_location = FedoraAuthzTests.getAclUri(r)
+ binary_description = self.find_binary_description(r)
+
+ self.assertNotEqual(binary_location, binary_description)
+
+ self.log("Check binary's acl link header is correct.")
+ self.checkValue(expected_acl, acl_location)
+
+ self.log("Check binary description's acl link header is correct")
+
+ r = self.do_get(binary_location)
+ self.checkResponse(200, r)
+ acl_location = FedoraAuthzTests.getAclUri(r)
+ self.checkValue(expected_acl, acl_location)
+
+ self.log("Check binary timemap's acl link header is correct.")
+ binary_versions = binary_location + "/fcr:versions"
+ r = self.do_get(binary_versions)
+ self.checkResponse(200, r)
+ acl_location = FedoraAuthzTests.getAclUri(r)
+ self.checkValue(expected_acl, acl_location)
+
+ self.log("Create a version of binary")
+ r = self.do_post(parent=binary_versions)
+ self.checkResponse(201, r)
+ memento_location = self.get_location(r)
+
+ self.log("Check memento's acl link header is correct")
+ r = self.do_get(memento_location)
+ self.checkResponse(200, r)
+ acl_location = FedoraAuthzTests.getAclUri(r)
+ self.checkValue(expected_acl, acl_location)
+
+ self.log("Check binary description timemap's acl link header is correct.")
+ binary_metadata_versions = binary_description + "/fcr:versions"
+ r = self.do_get(binary_metadata_versions)
+ self.checkResponse(200, r)
+ acl_location = FedoraAuthzTests.getAclUri(r)
+ self.checkValue(expected_acl, acl_location)
+
+ # metadata version is same datetime, so create it.
+ metadata_memento_location = binary_metadata_versions + memento_location[memento_location.rfind('/'):]
+
+ self.log("Check metadata memento's acl link header is correct")
+ r = self.do_get(metadata_memento_location)
+ self.checkResponse(200, r)
+ acl_location = FedoraAuthzTests.getAclUri(r)
+ self.checkValue(expected_acl, acl_location)
+
+ self.log("Try to get the ACL")
+ r = self.do_get(acl_location)
+ self.checkResponse(404, r)
+
+ @Test
+ def binaryAndMetadataShareACL(self):
+ self.verifyAuthEnabled()
+
+ self.log("Create a target binary")
+ headers = {
+ 'Content-type': 'text/plain',
+ 'Link': self.make_type(TestConstants.LDP_NON_RDF_SOURCE)
+ }
+ r = self.do_post(headers=headers, body="this is a test payload")
+ self.checkResponse(201, r)
+ binary_location = self.get_location(r)
+ acl_location = FedoraAuthzTests.getAclUri(r)
+ binary_description = self.find_binary_description(r)
+
+ self.log("Add ACL allowing write to metadata but not binary for user")
+ binary_acl = "@prefix acl: <{0}> .\n" \
+ "<#binary> a acl:Authorization ;\n" \
+ " acl:mode acl:Write ;\n" \
+ " acl:accessTo <{1}> ;\n" \
+ " acl:agent \"{2}\" .\n".format(TestConstants.ACL_NS, binary_location,
+ self.config[TestConstants.USER_NAME_PARAM])
+ headers = {
+ 'Content-type': TestConstants.TURTLE_MIMETYPE
+ }
+ r = self.do_put(acl_location, headers=headers, body=binary_acl)
+ self.checkResponse(201, r)
+
+ self.log("Try to read binary")
+ r = self.do_head(binary_location, admin=False)
+ self.checkResponse(403, r)
+
+ self.log("Try to write binary")
+ headers = {
+ 'Content-type': 'text/plain'
+ }
+ r = self.do_put(binary_location, headers=headers, body="This is a new body", admin=False)
+ self.checkResponse(204, r)
+
+ self.log("Try to read the metadata")
+ r = self.do_get(binary_description, admin=False)
+ self.checkResponse(403, r)
+
+ self.log("Try to patch metadata")
+ patch_body = "prefix dc: <{0}> INSERT DATA {{ <> dc:title \"Updated title\"}}".format(TestConstants.PURL_NS)
+ headers = {
+ 'Content-type': TestConstants.SPARQL_UPDATE_MIMETYPE
+ }
+ r = self.do_patch(binary_description, patch_body, headers=headers, admin=False)
+ self.checkResponse(204, r)
+
+ @Test
+ def testContainerWithAccessToClass(self):
+
+ self.verifyAuthEnabled()
+
+ self.log("Create a container")
+ r = self.do_post()
+ location = self.get_location(r)
+ acl_location = self.getAclUri(r)
+ versions_location = location + "/" + TestConstants.FCR_VERSIONS
+
+ full_acl = "@prefix acl: <{0}> .\n" \
+ "@prefix fedora: <{1}> .\n" \
+ "@prefix memento: <{2}> .\n" \
+ "<#container> a acl:Authorization ;\n" \
+ " acl:mode acl:Read ;\n" \
+ " acl:accessTo <{3}> ;\n" \
+ " acl:agent \"{4}\" .\n" \
+ "<#timemap> a acl:Authorization ;\n" \
+ " acl:mode acl:Read ;\n" \
+ " acl:accessToClass fedora:TimeMap ;\n" \
+ " acl:default <{3}> ;\n" \
+ " acl:agent \"{4}\" .\n" \
+ "<#memento> a acl:Authorization ;\n" \
+ " acl:mode acl:Read ;\n" \
+ " acl:accessToClass memento:Memento ;\n" \
+ " acl:default <{3}> ;\n" \
+ " acl:agent \"{5}\" .\n".format(TestConstants.ACL_NS, TestConstants.FEDORA_NS, TestConstants.MEMENTO_NS,
+ location, self.config[TestConstants.USER_NAME_PARAM],
+ self.config[TestConstants.ADMIN_USER_PARAM])
+ turtle_headers = {
+ 'Content-type': TestConstants.TURTLE_MIMETYPE
+ }
+
+ user2 = self.create_user2_auth()
+
+ self.log("Put ACL giving test user 1 read access to container and timemap but not mementos")
+ r = self.do_put(acl_location, headers=turtle_headers, body=full_acl)
+ self.checkResponse(201, r)
+
+ self.log("Create a Memento as admin")
+ r = self.do_post(versions_location)
+ self.checkResponse(201, r)
+ memento_location = self.get_location(r)
+
+ self.log("Try to get container as test user 1")
+ r = self.do_get(location, admin=False)
+ self.checkResponse(200, r)
+
+ self.log("Try to get container as test user 2")
+ r = self.do_get(location, admin=user2)
+ self.checkResponse(403, r)
+
+ self.log("Try to get timemap as test user 1")
+ r = self.do_get(versions_location, admin=False)
+ self.checkResponse(200, r)
+
+ self.log("Try to get timemap as test user 2")
+ r = self.do_get(versions_location, admin=user2)
+ self.checkResponse(403, r)
+
+ self.log("Check that memento exists as admin")
+ r = self.do_get(memento_location)
+ self.checkResponse(200, r)
+
+ self.log("Try to get memento as test user 1")
+ r = self.do_get(memento_location, admin=False)
+ self.checkResponse(403, r)
+
+ self.log("Try to get memento as test user 2")
+ r = self.do_get(memento_location, admin=user2)
+ self.checkResponse(403, r)
diff --git a/basic_interaction_tests.py b/basic_interaction_tests.py
index 4c270dc..f0c89c3 100644
--- a/basic_interaction_tests.py
+++ b/basic_interaction_tests.py
@@ -31,59 +31,43 @@ def createTestResource(self, type, files=None):
@Test
def testBasicContainer(self):
- self.log("Running testBasicContainer")
self.createTestResource(TestConstants.LDP_BASIC)
- self.log("Passed")
@Test
def testDirectContainer(self):
- self.log("Running testDirectContainer")
self.createTestResource(TestConstants.LDP_DIRECT)
- self.log("Passed")
@Test
def testIndirectContainer(self):
- self.log("Running testIndirectContainer")
self.createTestResource(TestConstants.LDP_INDIRECT)
- self.log("Passed")
@Test
def testNonRdfSource(self):
- self.log("Running testNonRdfSource")
testfiles = {'files': ('testdata.csv', 'this,is,some,data\n')}
self.createTestResource(TestConstants.LDP_NON_RDF_SOURCE, files=testfiles)
- self.log("Passed")
@Test
def testLdpResource(self):
""" We don't allow you to create a ldp:Resource so this returns 400 Bad Request """
- self.log("Running testLdpResource")
link_type = self.make_type(TestConstants.LDP_RESOURCE)
headers = {
'Link': link_type
}
r = self.do_post(self.getBaseUri(), headers=headers)
self.assertEqual(400, r.status_code, "Did not get expected response")
- self.log("Passed")
@Test
def testLdpContainer(self):
""" We don't allow you to create a ldp:Container so this returns 400 Bad Request """
- self.log("Running testLdpResource")
link_type = self.make_type(TestConstants.LDP_CONTAINER)
headers = {
'Link': link_type
}
r = self.do_post(self.getBaseUri(), headers=headers)
self.assertEqual(400, r.status_code, "Did create container")
- self.log("Passed")
-
-
@Test
def doNestedTests(self):
- self.log("Running doNestedTests")
-
self.log("Create a container")
headers = {
'Content-type': 'text/turtle'
@@ -125,8 +109,6 @@ def doNestedTests(self):
r = self.do_get(location)
self.assertEqual(410, r.status_code, "Did not get expected status code")
- self.log("Passed")
-
def changeIxnModels(self, location, starting_model):
""" This function uses a created object at {location} with starting type {starting_model}.
The below dictionary of tuples works as such
@@ -179,12 +161,8 @@ def changeIxnModels(self, location, starting_model):
r = self.do_put(location, headers=headers, files=files)
self.assertEqual(result, r.status_code, "Did not get expected response")
- self.log("Passed")
-
@Test
def testChangeIxnModel(self):
- self.log("Running changeIxnModel")
-
self.log("Create a basic container")
basic = self.createTestResource(TestConstants.LDP_BASIC)
self.changeIxnModels(basic, TestConstants.LDP_BASIC)
diff --git a/camel_tests.py b/camel_tests.py
new file mode 100644
index 0000000..d85716b
--- /dev/null
+++ b/camel_tests.py
@@ -0,0 +1,136 @@
+#!/bin/env python
+
+import TestConstants
+from abstract_fedora_tests import FedoraTests, register_tests, Test
+import uuid
+import datetime
+import requests
+import json
+import pyjq
+
+
+@register_tests
+class FedoraCamelTests(FedoraTests):
+
+ # Create test objects all inside here for easy of review.
+ CONTAINER = "/test_camel"
+
+ # How many seconds to wait for indexing to Solr and/or triplestore.
+ CONTAINER_WAIT = 30
+
+ def run_tests(self):
+ if not (self.hasSolrUrl() and self.hasTriplestoreUrl()):
+ print("**** Cannot run camel tests without a Solr and/or Triplestore base url ****")
+ else:
+ super().run_tests()
+
+ @Test
+ def CamelCreateObject(self):
+ self.log("Create an object")
+ internal_id = str(uuid.uuid4())
+ expected_url = self.getBaseUri() + "/" + internal_id
+ headers = {
+ 'Slug': internal_id,
+ 'Content-type': TestConstants.TURTLE_MIMETYPE
+ }
+ r = self.do_post(headers=headers, body=TestConstants.PCDM_CONTAINER_TTL)
+ self.assertEqual(201, r.status_code, "Did not get expected response code")
+
+ if self.hasSolrUrl():
+ self.log("Checking for item in Solr")
+ solr_response = self.timedQuery('querySolr', 'solrNumFound', expected_url)
+ if solr_response is not None:
+ self.assertEqual(expected_url, self.getIdFromSolr(solr_response), "Did not find ID in Solr.")
+ else:
+ self.fail("Timed out waiting to find record in Solr.")
+
+ if self.hasTriplestoreUrl():
+ self.log("Checking for item in Triplestore")
+ triplestore_response = self.timedQuery('queryTriplestore', 'triplestoreNumFound', expected_url)
+ if triplestore_response is not None:
+ self.assertEqual(TestConstants.PCDM_CONTAINER_TITLE, self.getIdFromTriplestore(triplestore_response),
+ "Did not find ID in Triplestore")
+ else:
+ self.fail("Timed out waiting to find record in Triplestore")
+
+ # Utility functions.
+ def timedQuery(self, query_method, check_method, query_value):
+ query_func = getattr(self, query_method, None)
+ check_func = getattr(self, check_method, None)
+ if query_func is None or check_func is None:
+ Exception("Can't find expected methods {0}, {1}".format(query_method, check_method))
+ current_time = datetime.datetime.now()
+ end_time = current_time + datetime.timedelta(seconds=self.CONTAINER_WAIT)
+ last_query = None
+ while current_time <= end_time:
+ # If a multiple of 5 seconds has passed since the last query, do the query
+ if last_query is None or (current_time - last_query).seconds >= 5:
+ last_query = datetime.datetime.now()
+ response = query_func(query_value)
+ if check_func(response) is not None:
+ return response
+ current_time = datetime.datetime.now()
+ return None
+
+ def hasSolrUrl(self):
+ try:
+ return self.config[TestConstants.SOLR_URL_PARAM] is not None and \
+ len(self.config[TestConstants.SOLR_URL_PARAM].strip()) > 0
+ except KeyError:
+ return False
+
+ def hasTriplestoreUrl(self):
+ try:
+ return self.config[TestConstants.TRIPLESTORE_URL_PARAM] is not None and \
+ len(self.config[TestConstants.TRIPLESTORE_URL_PARAM].strip()) > 0
+ except KeyError:
+ return False
+
+ def solrNumFound(self, response):
+ body = response.content.decode('UTF-8')
+ json_body = json.loads(body)
+ num_found = pyjq.first('.response.numFound', json_body)
+ if num_found is not None and int(num_found) > 0:
+ return num_found
+ return None
+
+ def querySolr(self, expected_id):
+ solr_select = self.config[TestConstants.SOLR_URL_PARAM].rstrip('/') + "/select"
+ params = {
+ 'q': 'id:"' + expected_id + '"',
+ 'wt': 'json'
+ }
+ r = requests.get(solr_select, params=params)
+ self.assertEqual(200, r.status_code, "Did not query Solr successfully.")
+ return r
+
+ def getIdFromSolr(self, response):
+ body = response.content.decode('UTF-8')
+ json_body = json.loads(body)
+ found_title = pyjq.first('.response.docs[].id', json_body)
+ return found_title
+
+ def queryTriplestore(self, expected_id):
+ query = "PREFIX dc: SELECT ?o WHERE { <" + expected_id + "> dc:title ?o}"
+ headers = {
+ 'Content-type': TestConstants.SPARQL_QUERY_MIMETYPE,
+ 'Accept': TestConstants.SPARQL_RESULT_JSON_MIMETYPE
+ }
+ r = requests.post(self.config[TestConstants.TRIPLESTORE_URL_PARAM], headers=headers, data=query)
+ self.assertEqual(200, r.status_code, 'Did not query Triplestore successfully.')
+ return r
+
+ def triplestoreNumFound(self, response):
+ body = response.content.decode('UTF-8')
+ json_body = json.loads(body)
+ # This results a list of matching bindings, so it can be an empty list.
+ num_found = pyjq.all('.results.bindings[].o', json_body)
+ if num_found is not None and len(num_found) > 0:
+ return num_found
+ return None
+
+ def getIdFromTriplestore(self, response):
+ body = response.content.decode('UTF-8')
+ json_body = json.loads(body)
+ found_id = pyjq.first('.results.bindings[].o.value', json_body)
+ return found_id
diff --git a/config.yml.example b/config.yml.example
index dc7b0fc..7be2a77 100644
--- a/config.yml.example
+++ b/config.yml.example
@@ -7,3 +7,13 @@ default:
password1: testpass
user2: testuser2
password2: testpass
+tomcat:
+ baseurl: http://localhost:8080/fcrepo/rest
+ admin_user: fedoraAdmin
+ admin_password: secret3
+ user1: adminuser
+ password1: password2
+ user2: testuser
+ password2: password1
+ solrurl: http://localhost:8080/solr
+ triplestoreurl: http://localhost:8080/fuseki/test/query
diff --git a/fixity_tests.py b/fixity_tests.py
index 30d4244..4f6e9ed 100644
--- a/fixity_tests.py
+++ b/fixity_tests.py
@@ -43,5 +43,3 @@ def aFixityTest(self):
'."http://www.loc.gov/premis/rdf/v1#hasMessageDigest" | .[]?."@id"'.format(fixity_id),
json_body)
self.assertEqual(self.FIXITY_RESULT, fixity_result, "Fixity result was not a match for expected.")
-
- self.log("Passed")
\ No newline at end of file
diff --git a/indirect_tests.py b/indirect_tests.py
index bd9fcea..0389327 100644
--- a/indirect_tests.py
+++ b/indirect_tests.py
@@ -14,8 +14,6 @@ class FedoraIndirectTests(FedoraTests):
@Test
def doPcdmIndirect(self):
- self.log("Running doPcdmIndirect")
-
self.log("Create a PCDM container")
basic_headers = {
'Link': self.make_type(TestConstants.LDP_BASIC),
@@ -70,5 +68,3 @@ def doPcdmIndirect(self):
if member == pcdm_container_location:
found_member = True
self.assertTrue(found_member, "Did not find hasMember property")
-
- self.log("Passed")
\ No newline at end of file
diff --git a/rdf_tests.py b/rdf_tests.py
index dc65436..3e9959d 100644
--- a/rdf_tests.py
+++ b/rdf_tests.py
@@ -20,8 +20,6 @@ class FedoraRdfTests(FedoraTests):
@Test
def testRdfSerialization(self):
- self.log("Starting testRdfSerialization")
-
self.log("Put new resource.")
headers = {
"Content-type": "text/turtle"
@@ -63,5 +61,3 @@ def testRdfSerialization(self):
self.log("Test for tombstone")
r = self.do_get(location)
self.assertEqual(410, r.status_code, "Object's tombstone not found.")
-
- self.log("Passed")
\ No newline at end of file
diff --git a/sparql_tests.py b/sparql_tests.py
index 4a0878c..35d297e 100644
--- a/sparql_tests.py
+++ b/sparql_tests.py
@@ -19,8 +19,6 @@ class FedoraSparqlTests(FedoraTests):
@Test
def doSparqlContainerTest(self):
- self.log("Running doSparqlContainerTest")
-
self.log("Create container")
headers = {
'Content-type': 'text/turtle'
@@ -42,12 +40,8 @@ def doSparqlContainerTest(self):
self.assertEqual(204, r.status_code, "Did not get expected results")
self.assertTitleExists("Updated title", location)
- self.log("Passed")
-
@Test
def doSparqlBinaryTest(self):
- self.log("Running doSparqlBinaryTest")
-
self.log("Create a binary")
headers = {
'Content-type': 'image/jpeg'
@@ -71,12 +65,8 @@ def doSparqlBinaryTest(self):
self.assertEqual(204, r.status_code, "Did not get expected results")
self.assertTitleExists("Updated title", description)
- self.log("Passed")
-
@Test
def doUnicodeSparql(self):
- self.log("Running doUnicodeSparql")
-
self.log("Create a container")
r = self.do_post(self.getBaseUri())
self.assertEqual(201, r.status_code, "Did not get expected response code")
@@ -93,5 +83,3 @@ def doUnicodeSparql(self):
r = self.do_patch(location, headers=headers, body=sparql)
self.assertEqual(204, r.status_code, "Did not get expected response code")
self.assertTitleExists("Die von Blumenbach gegründete anthropologische Sammlung der Universität", location)
-
- self.log("Passed")
\ No newline at end of file
diff --git a/testrunner.py b/testrunner.py
index f7baf14..aa19dc6 100755
--- a/testrunner.py
+++ b/testrunner.py
@@ -17,6 +17,7 @@
from transaction_tests import FedoraTransactionTests
from authz_tests import FedoraAuthzTests
from indirect_tests import FedoraIndirectTests
+from camel_tests import FedoraCamelTests
class FedoraTestRunner:
@@ -31,7 +32,9 @@ class FedoraTestRunner:
(TestConstants.USER2_NAME_PARAM, True),
(TestConstants.USER2_PASS_PARAM, True),
(TestConstants.LOG_FILE_PARAM, False),
- (TestConstants.SELECTED_TESTS_PARAM, False)
+ (TestConstants.SELECTED_TESTS_PARAM, False),
+ (TestConstants.SOLR_URL_PARAM, False),
+ (TestConstants.TRIPLESTORE_URL_PARAM, False)
]
config = {}
logger = None
@@ -99,6 +102,9 @@ def run_tests(self):
if test == 'all' or test == 'indirect':
indirect = FedoraIndirectTests(self.config)
indirect.run_tests()
+ if test == 'camel':
+ camel = FedoraCamelTests(self.config)
+ camel.run_tests()
def main(self, args):
self.set_up(args)
@@ -116,7 +122,7 @@ def csv_list(string):
class CSVAction(argparse.Action):
- valid_options = ["authz", "basic", "sparql", "rdf", "version", "transaction", "fixity", "indirect"]
+ valid_options = ["authz", "basic", "sparql", "rdf", "version", "transaction", "fixity", "indirect", "camel"]
def __call__(self, parser, args, values, option_string=None):
if isinstance(values, list):
@@ -158,7 +164,7 @@ def __call__(self, parser, args, values, option_string=None):
parser.add_argument('-k', '--' + TestConstants.USER2_PASS_PARAM, dest=TestConstants.USER2_PASS_PARAM,
help="Second regular user password")
parser.add_argument('-t', '--tests', dest="selected_tests", help='Comma separated list of which tests to run from '
- '{0}. Defaults to running all tests'.format(",".join(CSVAction.valid_options)),
+ '{0}. Defaults to running all tests'.format(", ".join(CSVAction.valid_options)),
default=['all'], type=csv_list, action=CSVAction)
args = parser.parse_args()
diff --git a/transaction_tests.py b/transaction_tests.py
index d271c8e..f01e0a8 100644
--- a/transaction_tests.py
+++ b/transaction_tests.py
@@ -26,8 +26,6 @@ def get_transaction_provider(self):
@Test
def doCommitTest(self):
- self.log("Running doCommitTest")
-
tx_provider = self.get_transaction_provider()
if tx_provider is None:
self.log("Could not location transaction provider")
@@ -66,12 +64,8 @@ def doCommitTest(self):
r = self.do_get(outside_location)
self.assertEqual(200, r.status_code, "Did not get expected response code")
- self.log("Passed")
-
@Test
def doRollbackTest(self):
- self.log("Running doRollbackTest")
-
tx_provider = self.get_transaction_provider()
if tx_provider is None:
self.log("Could not location transaction provider")
@@ -105,5 +99,3 @@ def doRollbackTest(self):
self.log("Container is still not available outside the transaction")
r = self.do_get(outside_location)
self.assertEqual(404, r.status_code, "Did not get expected response code")
-
- self.log("Passed")
diff --git a/version_tests.py b/version_tests.py
index f6e1e4e..25f78d9 100644
--- a/version_tests.py
+++ b/version_tests.py
@@ -17,55 +17,67 @@ def count_mementos(response):
mementos = [x for x in body.split('\n') if x.find("rel=\"memento\"") >= 0]
return len(mementos)
+ def checkMementoCount(self, expected, uri, admin=None):
+ if admin is None:
+ admin = True
+ if not uri.endswith("/" + TestConstants.FCR_VERSIONS):
+ uri += "/" + TestConstants.FCR_VERSIONS
+ headers = {
+ 'Accept': TestConstants.LINK_FORMAT_MIMETYPE
+ }
+ r = self.do_get(uri, headers=headers, admin=admin)
+ self.checkResponse(200, r)
+ self.checkValue(expected, self.count_mementos(r))
+
@Test
def doContainerVersioningTest(self):
- self.log("Running doContainerVersioningTest")
headers = {
'Link': self.make_type(TestConstants.LDP_BASIC)
}
r = self.do_post(self.getBaseUri(), headers=headers)
self.log("Create a basic container")
- self.assertEqual(201, r.status_code, "Did not create container")
+ self.checkResponse(201, r)
location = self.get_location(r)
version_endpoint = location + "/" + TestConstants.FCR_VERSIONS
r = self.do_get(version_endpoint)
- self.assertEqual(200, r.status_code, "Did not find versions where we should")
+ self.checkResponse(200, r)
self.log("Create a version")
r = self.do_post(version_endpoint)
- self.assertEqual(201, r.status_code, 'Did not create a version')
+ self.checkResponse(201, r)
self.log("Get the resource content")
headers = {
'Accept': TestConstants.JSONLD_MIMETYPE
}
r = self.do_get(location, headers=headers)
- self.assertEqual(200, r.status_code, "Could not get the resource")
+ self.checkResponse(200, r)
body = r.content.decode('UTF-8').rstrip('\n')
new_date = FedoraTests.get_rfc_date("2000-06-01 08:21:00")
- self.log("Create a version with provided datetime")
headers = {
'Content-Type': TestConstants.JSONLD_MIMETYPE,
'Prefer': TestConstants.PUT_PREFER_LENIENT,
'Memento-Datetime': new_date
}
+ self.log("Check you must provide a valid body")
r = self.do_post(version_endpoint, headers=headers, body="[]")
- self.assertEqual(400, r.status_code, "Empty body should cause Bad request")
+ self.checkResponse(400, r)
+ self.log("Create a version with provided datetime")
r = self.do_post(version_endpoint, headers=headers, body=body)
- self.assertEqual(201, r.status_code, "Did not create memento will body.")
+ self.checkResponse(201, r)
memento_location = self.get_location(r)
self.log("Try creating another version at the same location")
r = self.do_post(version_endpoint, headers=headers, body=body)
- self.assertEqual(409, r.status_code, "Did not get conflict trying to create a second memento")
+ self.checkResponse(409, r)
self.log("Check memento exists")
r = self.do_get(memento_location)
- self.assertEqual(200, r.status_code, "Memento was not found")
+ self.checkResponse(200, r)
found_datetime = r.headers['Memento-Datetime']
self.assertIsNotNone(found_datetime, "Did not find Memento-Datetime header")
self.assertEqual(0, self.compare_rfc_dates(new_date, found_datetime), "Returned Memento-Datetime did not match sent")
@@ -79,39 +91,32 @@ def doContainerVersioningTest(self):
"Content-Type": TestConstants.SPARQL_UPDATE_MIMETYPE
}
r = self.do_patch(location, headers=headers, body=sparql_body)
- self.assertEqual(204, r.status_code, "Unable to patch")
+ self.checkResponse(204, r)
self.log("Try to patch the memento")
r = self.do_patch(memento_location, headers=headers, body=sparql_body)
- self.assertEqual(405, r.status_code, "Did not get denied PATCH request.")
+ self.checkResponse(405, r)
- # Delay one second to ensure we don't POST at the same time
+ self.log("Wait a second to change the time.")
time.sleep(1)
self.log("Create another version")
r = self.do_post(version_endpoint)
- self.assertEqual(201, r.status_code, 'Did not create a version')
+ self.checkResponse(201, r)
self.log("Count mementos")
- headers = {
- 'Accept': 'application/link-format'
- }
- r = self.do_get(version_endpoint, headers=headers)
- self.assertEqual(200, r.status_code, "Did not get the fcr:versions content")
- self.assertEqual(3, FedoraVersionTests.count_mementos(r), "Incorrect number of Mementos")
+ self.checkMementoCount(3, version_endpoint)
self.log("Delete a Memento")
r = self.do_delete(memento_location)
- self.assertEqual(204, r.status_code, "Did not delete the Memento")
+ self.checkResponse(204, r)
self.log("Validate delete")
r = self.do_get(memento_location)
- self.assertEqual(404, r.status_code, "Memento was not gone")
+ self.checkResponse(404, r)
self.log("Validate delete with another count")
- r = self.do_get(version_endpoint, headers=headers)
- self.assertEqual(200, r.status_code, "Did not get the fcr:versions content")
- self.assertEqual(2, FedoraVersionTests.count_mementos(r), "Incorrect number of Mementos")
+ self.checkMementoCount(2, version_endpoint)
self.log("Create a memento at the deleted datetime")
headers = {
@@ -120,18 +125,14 @@ def doContainerVersioningTest(self):
'Memento-Datetime': new_date
}
r = self.do_post(version_endpoint, headers=headers, body=body)
- self.assertEqual(201, r.status_code, "Did not create version with specific date and body")
+ self.checkResponse(201, r)
self.log("Check the memento exists again")
r = self.do_get(memento_location)
- self.assertEqual(200, r.status_code, "Memento was not found")
-
- self.log("Passed")
+ self.checkResponse(200, r)
@Test
def doBinaryVersioningTest(self):
- self.log("Running doBinaryVersioningTest")
-
headers = {
'Link': self.make_type(TestConstants.LDP_NON_RDF_SOURCE),
'Content-Type': 'text/csv'
@@ -141,16 +142,34 @@ def doBinaryVersioningTest(self):
r = self.do_post(self.getBaseUri(), headers=headers, files=files)
self.log("Create a NonRdfSource")
- self.assertEqual(201, r.status_code, "Did not create NonRdfSource")
+ self.checkResponse(201, r)
location = self.get_location(r)
+ description_location = self.find_binary_description(r)
version_endpoint = location + "/" + TestConstants.FCR_VERSIONS
+ description_version_endpoint = description_location + "/" + TestConstants.FCR_VERSIONS
+
+ self.log("Get version endpoint")
r = self.do_get(version_endpoint)
- self.assertEqual(200, r.status_code, "Did not find versions where we should")
+ self.checkResponse(200, r)
+
+ self.log("Get description version endpoint")
+ r = self.do_get(description_version_endpoint)
+ self.checkResponse(200, r)
self.log("Create a version")
r = self.do_post(version_endpoint)
- self.assertEqual(201, r.status_code, 'Did not create a version')
+ self.checkResponse(201, r)
+
+ self.log("Try to create another version too quickly")
+ r = self.do_post(version_endpoint)
+ self.checkResponse(409, r)
+
+ self.assertNotEqual(version_endpoint, description_version_endpoint)
+
+ self.log("Try to create a version of the description quickly")
+ r = self.do_post(description_version_endpoint)
+ self.checkResponse(409, r)
new_date = FedoraTests.get_rfc_date("2000-06-01 08:21:00")
@@ -160,16 +179,16 @@ def doBinaryVersioningTest(self):
'Memento-Datetime': new_date
}
r = self.do_post(version_endpoint, headers=headers, files=files)
- self.assertEqual(201, r.status_code, "Did not create version with specific date and body")
+ self.checkResponse(201, r)
memento_location = self.get_location(r)
self.log("Try creating another version at the same location")
r = self.do_post(version_endpoint, headers=headers, files=files)
- self.assertEqual(409, r.status_code, "Did not get conflict trying to create a second memento")
+ self.checkResponse(409, r)
self.log("Check memento exists")
r = self.do_head(memento_location)
- self.assertEqual(200, r.status_code, "Memento was not found")
+ self.checkResponse(200, r)
found_datetime = r.headers['Memento-Datetime']
self.assertIsNotNone(found_datetime, "Did not find Memento-Datetime header")
self.assertEqual(0, self.compare_rfc_dates(new_date, found_datetime), "Returned Memento-Datetime did not match sent")
@@ -180,39 +199,32 @@ def doBinaryVersioningTest(self):
'Content-Type': 'text/csv'
}
r = self.do_put(location, headers=headers, files=files)
- self.assertEqual(204, r.status_code, "Unable to put")
+ self.checkResponse(204, r)
self.log("Try to put to the memento")
r = self.do_put(memento_location, headers=headers, files=files)
- self.assertEqual(405, r.status_code, "Did not get denied PUT request.")
+ self.checkResponse(405, r)
- # Delay one second to ensure we don't POST at the same time
+ self.log("Wait one second to change the time.")
time.sleep(1)
self.log("Create another version")
r = self.do_post(version_endpoint)
- self.assertEqual(201, r.status_code, 'Did not create a version')
+ self.checkResponse(201, r)
self.log("Count mementos")
- headers = {
- 'Accept': 'application/link-format'
- }
- r = self.do_get(version_endpoint, headers=headers)
- self.assertEqual(200, r.status_code, "Did not get the fcr:versions content")
- self.assertEqual(3, FedoraVersionTests.count_mementos(r), "Incorrect number of Mementos")
+ self.checkMementoCount(3, version_endpoint)
self.log("Delete a Memento")
r = self.do_delete(memento_location)
- self.assertEqual(204, r.status_code, "Did not delete the Memento")
+ self.checkResponse(204, r)
self.log("Validate delete")
r = self.do_get(memento_location)
- self.assertEqual(404, r.status_code, "Memento was not gone")
+ self.checkResponse(404, r)
self.log("Validate delete with another count")
- r = self.do_get(version_endpoint, headers=headers)
- self.assertEqual(200, r.status_code, "Did not get the fcr:versions content")
- self.assertEqual(2, FedoraVersionTests.count_mementos(r), "Incorrect number of Mementos")
+ self.checkMementoCount(2, version_endpoint)
self.log("Create a memento at the deleted datetime")
headers = {
@@ -221,11 +233,122 @@ def doBinaryVersioningTest(self):
'Memento-Datetime': new_date
}
r = self.do_post(version_endpoint, headers=headers, files=files)
- self.assertEqual(201, r.status_code, "Did not create version with specific date and body")
+ self.checkResponse(201, r)
self.log("Check the memento exists again")
r = self.do_head(memento_location)
- self.assertEqual(200, r.status_code, "Memento was not found")
+ self.checkResponse(200, r)
+
+ self.log("Validate count of mementos again")
+ self.checkMementoCount(3, version_endpoint)
+
+ @Test
+ def checkBinaryVersioning(self):
+ headers = {
+ 'Link': self.make_type(TestConstants.LDP_NON_RDF_SOURCE),
+ 'Content-Type': 'text/csv'
+ }
+ files = {'file': ('report.csv', 'some,data,to,send\nanother,row,to,send\n')}
+
+ self.log("Create a NonRdfSource")
+ r = self.do_post(self.getBaseUri(), headers=headers, files=files)
+ self.checkResponse(201, r)
+ location = self.get_location(r)
+ description_location = self.find_binary_description(r)
+
+ binary_versions = location + "/" + TestConstants.FCR_VERSIONS
+ metadata_versions = description_location + "/" + TestConstants.FCR_VERSIONS
+
+ link_headers = {
+ 'Accept': TestConstants.LINK_FORMAT_MIMETYPE
+ }
+
+ self.log("Count Mementos of binary")
+ r = self.do_get(binary_versions, headers=link_headers)
+ self.checkResponse(200, r)
+ self.checkValue(0, self.count_mementos(r))
+
+ self.log("Count Mementos of binary description")
+ r = self.do_get(metadata_versions, headers=link_headers)
+ self.checkResponse(200, r)
+ self.checkValue(0, self.count_mementos(r))
+
+ self.log("Create version of binary from existing")
+ r = self.do_post(binary_versions)
+ self.checkResponse(201, r)
+
+ self.log("Count Mementos of binary")
+ self.checkMementoCount(1, binary_versions)
+
+ self.log("Count Mementos of binary description")
+ self.checkMementoCount(1, metadata_versions)
+
+ self.log("Wait a second")
+ time.sleep(1)
+
+ self.log("Create version of binary metadata from existing")
+ r = self.do_post(metadata_versions)
+ self.checkResponse(201, r)
+
+ self.log("Count Mementos of binary")
+ self.checkMementoCount(1, binary_versions)
+
+ self.log("Count Mementos of binary description")
+ self.checkMementoCount(2, metadata_versions)
+
+ @Test
+ def createBinaryVersionsAtSameTime(self):
+ headers = {
+ 'Link': self.make_type(TestConstants.LDP_NON_RDF_SOURCE),
+ 'Content-Type': 'text/csv'
+ }
+ files = {'file': ('report.csv', 'some,data,to,send\nanother,row,to,send\n')}
+
+ self.log("Create a NonRdfSource")
+ r = self.do_post(self.getBaseUri(), headers=headers, files=files)
+ self.checkResponse(201, r)
+ location = self.get_location(r)
+ description_location = self.find_binary_description(r)
+
+ r = self.do_get(description_location)
+ new_body = "@prefix dc: <{0}> .\n".format(TestConstants.PURL_NS) + \
+ r.text[0:-2] + ";\n dc:title \"New title\" .\n".format(TestConstants.PURL_NS)
+
+ version_endpoint = location + "/" + TestConstants.FCR_VERSIONS
+ description_version_endpoint = description_location + "/" + TestConstants.FCR_VERSIONS
+
+ self.log("Check we have no mementos of binary or description")
+ self.checkMementoCount(0, version_endpoint)
+ self.checkMementoCount(0, description_version_endpoint)
+
+ the_date = FedoraTests.get_rfc_date('2019-05-21 18:30:00')
+ files = {'file': ('report.csv', 'some,data,to,send\nanother,row,to,send\nevent,more,data,tosend\n')}
+ headers = {
+ 'Content-Type': 'text/csv',
+ 'Memento-Datetime': the_date
+ }
+
+ self.log("Make version for {0} of binary".format(the_date))
+ r = self.do_post(version_endpoint, headers=headers, files=files)
+ self.checkResponse(201, r)
+
+ self.log("Check we only made a memento of the binary")
+ self.checkMementoCount(1, version_endpoint)
+ self.log("Check we still have no description mementos")
+ self.checkMementoCount(0, description_version_endpoint)
+
+ headers = {
+ 'Content-type': TestConstants.TURTLE_MIMETYPE,
+ 'Memento-Datetime': the_date,
+ 'Prefer': TestConstants.PUT_PREFER_LENIENT
+ }
+
+ self.log("Make version for {0} of binary description".format(the_date))
+ r = self.do_post(description_version_endpoint, headers=headers, body=new_body)
+ self.checkResponse(201, r)
- self.log("Passed")
+ self.log("Count binary mementos")
+ self.checkMementoCount(1, version_endpoint)
+ self.log("Count binary description mementos")
+ self.checkMementoCount(1, description_version_endpoint)