add method to upload sources from schema and metadata

src/pycrunch/elements.py

@@ -283,6 +283,7 @@ class ElementResponseHandler(lemonpy.ResponseHandler):
     parsers = {"application/json": parse_json_element_from_response}
 
     def status_401(self, r):
+        print(r.json())
         login_url = r.json()["urls"]["login_url"]
         if r.request.url == login_url:
             # This means that the requests was made on the login_url, so it was

src/pycrunch/importing.py

@@ -52,6 +52,27 @@ def wait_for_batch_status(self, batch, status):
         else:
             raise ValueError("The batch did not reach the '%s' state in the "
                              "given time. Please check again later." % status)
+
+    def add_schema_metadata(self, ds, source_url, filename, fp, mimetype, schema, metadata):
+        response = ds.session.post(
+            source_url, 
+            files={
+                "uploaded_file": (filename, fp, mimetype)
+            },
+            data={
+                "schema": schema, 
+                "metadata": metadata,
+                "crunchlake": "create",
+                "dataset_id": ds.get("self", "").split("/")[-2]
+            }
+        )
+
+        return shoji.Entity(ds.session, body={
+                "status_code": response.status_code,
+                "payload": response.payload, 
+                "source_url": response.headers.get("Location")
+            }
+        )
 
     def add_source(self, ds, filename, fp, mimetype):
         """Create a new Source on the given dataset and return its URL."""

tests/test_lake.py

@@ -0,0 +1,20 @@
+import json
+import subprocess
+import time
+
+
+class TestCrunchLakeWorkflow:
+    def test_ping(self):
+        command = ["cr.core.launch", "ping", "crunch-lake-ping", '--kwargs={"pong": "test-ping"}']
+        p = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        start = time.time()
+        stdout = p.stdout.readline()
+        stderr = p.stderr.readline()
+
+        while time.time() - start < 30:
+            try:
+                assert "Workflow launched with id" in stderr.decode("utf-8")
+            except AssertionError as assertion_error:
+                if not time.time() - start < 30:
+                    raise assertion_error
+            time.sleep(.01)