Add tempest tests for get_archive/put_archive

Change-Id: I3e0e3e344892f49b2626e5234cfc1275a61959d0
Closes-Bug: #1676061

zun_tempest_plugin/tests/tempest/api/clients.py

@@ -278,6 +278,16 @@ class ZunClient(rest_client.RestClient):
                                params=params),
             None, **kwargs)
 
+    def put_archive(self, container_id, params=None, **kwargs):
+        return self.post(
+            self.container_uri(container_id, action='put_archive',
+                               params=params), **kwargs)
+
+    def get_archive(self, container_id, params=None, **kwargs):
+        return self.get(
+            self.container_uri(container_id, action='get_archive',
+                               params=params), None, **kwargs)
+
 
 @contextlib.contextmanager
 def docker_client(docker_auth_url):
@@ -362,3 +372,8 @@ class DockerClient(object):
                        docker_auth_url=CONF.docker.api_url):
         with docker_client(docker_auth_url) as docker:
             return docker.remove_network(name)
+
+    def get_archive(self, container_id, path,
+                    docker_auth_url=CONF.docker.api_url):
+        with docker_client(docker_auth_url) as docker:
+            return docker.get_archive(container_id, path)

zun_tempest_plugin/tests/tempest/api/test_containers.py

@@ -10,8 +10,12 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+from io import BytesIO
+import tarfile
+import time
 import types
 
+from oslo_serialization import jsonutils as json
 from oslo_utils import encodeutils
 import six
 from tempest.lib.common.utils import data_utils
@@ -542,6 +546,43 @@ class TestContainer(base.BaseZunTest):
             model.uuid, params={'network': network})
         self._ensure_network_attached(model, network)
 
+    @decorators.idempotent_id('037d800c-2262-4e15-90cd-95292b5ef958')
+    def test_put_and_get_archive_from_container(self):
+        _, model = self._run_container()
+        self.assertEqual(1, len(model.addresses))
+
+        # Create a simple tarstream
+        file_content = 'Random text'
+
+        tarstream = BytesIO()
+        with tarfile.open(fileobj=tarstream, mode='w') as tar:
+            encoded_file_content = file_content.encode()
+            tarinfo = tarfile.TarInfo(name='test.txt')
+            tarinfo.size = len(encoded_file_content)
+            tarinfo.mtime = time.time()
+            tar.addfile(tarinfo, BytesIO(encoded_file_content))
+
+        # We're at the end of the tarstream, go back to the beginning
+        tarstream.seek(0)
+
+        req_body = json.dump_as_bytes({'data': tarstream.getvalue()})
+        resp, _ = self.container_client.put_archive(
+            model.uuid, params={'path': '/tmp'}, body=req_body)
+        self.assertEqual(200, resp.status)
+        resp, body = self.container_client.get_archive(
+            model.uuid, params={'path': '/tmp/test.txt'})
+        self.assertEqual(200, resp.status)
+
+        # Get content
+        body = json.loads(body)
+        tardata = BytesIO(body['data'].encode())
+        with tarfile.open(fileobj=tardata, mode='r') as tar:
+            untar_content = tar.extractfile('test.txt').read()
+
+        self.assertEqual(file_content, untar_content.decode())
+        self.assertEqual(body['stat']['name'], tarinfo.name)
+        self.assertEqual(body['stat']['size'], tarinfo.size)
+
     def _ensure_network_detached(self, container, network):
         def is_network_detached():
             _, model = self.container_client.get_container(container.uuid)