API for file upload. #116

system/api_lib.py

@@ -3,6 +3,9 @@ import time
 import json
 import random
 import string
+import os
+import inspect
+import shutil
 
 try:
     import requests
@@ -27,6 +30,9 @@ class APITest(BaseTest):
             APITest.aptly_server = self._start_process("aptly api serve -listen=%s" % (self.base_url),)
             time.sleep(1)
 
+        if os.path.exists(os.path.join(os.environ["HOME"], ".aptly", "upload")):
+            shutil.rmtree(os.path.join(os.environ["HOME"], ".aptly", "upload"))
+
     def run(self):
         pass
 
@@ -44,6 +50,23 @@ class APITest(BaseTest):
     def delete(self, uri, *args, **kwargs):
         return requests.delete("http://%s%s" % (self.base_url, uri), *args, **kwargs)
 
+    def upload(self, uri, *filenames, **kwargs):
+        upload_name = kwargs.pop("upload_name", None)
+        directory = kwargs.pop("directory", "files")
+        assert kwargs == {}
+
+        files = {}
+
+        for filename in filenames:
+            fp = open(os.path.join(os.path.dirname(inspect.getsourcefile(BaseTest)), directory, filename), "rb")
+            if upload_name is not None:
+                upload_filename = upload_name
+            else:
+                upload_filename = filename
+            files[upload_filename] = (upload_filename, fp)
+
+        return self.post(uri, files=files)
+
     @classmethod
     def shutdown_class(cls):
         if cls.aptly_server is not None:

system/t12_api/__init__.py

@@ -3,3 +3,4 @@ Testing aptly REST API
 """
 
 from .repos import *
+from .files import *

system/t12_api/files.py

@@ -0,0 +1,98 @@
+from api_lib import APITest
+
+
+class FilesAPITestUpload(APITest):
+    """
+    POST /files/:dir
+    """
+
+    def check(self):
+        d = self.random_name()
+        resp = self.upload("/api/files/" + d, "pyspi_0.6.1-1.3.dsc")
+        self.check_equal(resp.status_code, 200)
+        self.check_equal(resp.json(), [d + '/pyspi_0.6.1-1.3.dsc'])
+        self.check_exists("upload/" + d + '/pyspi_0.6.1-1.3.dsc')
+
+
+class FilesAPITestUploadMulti(APITest):
+    """
+    POST /files/:dir, GET /files/:dir multi files
+    """
+
+    def check(self):
+        d = self.random_name()
+
+        self.check_equal(self.get("/api/files/" + d).status_code, 404)
+
+        resp = self.upload("/api/files/" + d, "pyspi_0.6.1-1.3.dsc", "pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz")
+        self.check_equal(resp.status_code, 200)
+        self.check_equal(sorted(resp.json()),
+                         [d + '/pyspi_0.6.1-1.3.diff.gz', d + '/pyspi_0.6.1-1.3.dsc', d + '/pyspi_0.6.1.orig.tar.gz'])
+        self.check_exists("upload/" + d + '/pyspi_0.6.1-1.3.dsc')
+        self.check_exists("upload/" + d + '/pyspi_0.6.1-1.3.diff.gz')
+        self.check_exists("upload/" + d + '/pyspi_0.6.1.orig.tar.gz')
+
+        resp = self.get("/api/files/" + d)
+        self.check_equal(resp.status_code, 200)
+        self.check_equal(sorted(resp.json()),
+                         ['pyspi_0.6.1-1.3.diff.gz', 'pyspi_0.6.1-1.3.dsc', 'pyspi_0.6.1.orig.tar.gz'])
+
+
+class FilesAPITestList(APITest):
+    """
+    GET /files/
+    """
+
+    def check(self):
+        d1, d2, d3 = self.random_name(), self.random_name(), self.random_name()
+
+        resp = self.get("/api/files")
+        self.check_equal(resp.status_code, 200)
+        self.check_equal(resp.json(), [])
+
+        self.check_equal(self.upload("/api/files/" + d1, "pyspi_0.6.1-1.3.dsc").status_code, 200)
+        self.check_equal(self.upload("/api/files/" + d2, "pyspi_0.6.1-1.3.dsc").status_code, 200)
+        self.check_equal(self.upload("/api/files/" + d3, "pyspi_0.6.1-1.3.dsc").status_code, 200)
+
+        resp = self.get("/api/files")
+        self.check_equal(resp.status_code, 200)
+        self.check_equal(sorted(resp.json()), sorted([d1, d2, d3]))
+
+
+class FilesAPITestDelete(APITest):
+    """
+    DELETE /files/:dir, DELETE /files/:dir/:name
+    """
+
+    def check(self):
+        d1, d2 = self.random_name(), self.random_name()
+
+        self.check_equal(self.get("/api/files").json(), [])
+        self.check_equal(self.delete("/api/files/" + d1).status_code, 200)
+        self.check_equal(self.delete("/api/files/" + d1 + "/" + "pyspi_0.6.1-1.3.dsc").status_code, 200)
+
+        self.check_equal(self.upload("/api/files/" + d1, "pyspi_0.6.1-1.3.dsc").status_code, 200)
+        self.check_equal(self.upload("/api/files/" + d2, "pyspi_0.6.1-1.3.dsc").status_code, 200)
+
+        self.check_equal(self.delete("/api/files/" + d1).status_code, 200)
+        self.check_equal(self.get("/api/files").json(), [d2])
+
+        self.check_equal(self.delete("/api/files/" + d2 + "/" + "no-such-file").status_code, 200)
+        self.check_equal(self.get("/api/files/" + d2).json(), ["pyspi_0.6.1-1.3.dsc"])
+
+        self.check_equal(self.delete("/api/files/" + d2 + "/" + "pyspi_0.6.1-1.3.dsc").status_code, 200)
+        self.check_equal(self.get("/api/files").json(), [d2])
+        self.check_equal(self.get("/api/files/" + d2).json(), [])
+
+
+class FilesAPITestSecurity(APITest):
+    """
+    delete & upload security
+    """
+
+    def check(self):
+        self.check_equal(self.delete("/api/files/.").status_code, 400)
+        self.check_equal(self.delete("/api/files").status_code, 404)
+        self.check_equal(self.delete("/api/files/../.").status_code, 400)
+        self.check_equal(self.delete("/api/files/./..").status_code, 400)
+        self.check_equal(self.delete("/api/files/dir/..").status_code, 400)