KaneAI - World’s First E2E Software Testing Agent.Signup for Private Beta Join TestMu Conference

How to use _upload_archive method in tempest

Best Python code snippet using tempest_python

test_account_bulk.py

Source:test_account_bulk.py

...38        tar.add(tmp_dir, arcname=container_name)39        tar.close()40        tarpath.flush()41        return tarpath.name, container_name, object_name42    def _upload_archive(self, filepath):43        # upload an archived file44        params = {'extract-archive': 'tar'}45        with open(filepath) as fh:46            mydata = fh.read()47            resp, body = self.account_client.create_account(data=mydata,48                                                            params=params)49        return resp, body50    def _check_contents_deleted(self, container_name):51        param = {'format': 'txt'}52        resp, body = self.account_client.list_account_containers(param)53        self.assertHeaders(resp, 'Account', 'GET')54        self.assertNotIn(container_name, body)55    @test.idempotent_id('a407de51-1983-47cc-9f14-47c2b059413c')56    @test.requires_ext(extension='bulk', service='object')57    def test_extract_archive(self):58        # Test bulk operation of file upload with an archived file59        filepath, container_name, object_name = self._create_archive()60        resp, _ = self._upload_archive(filepath)61        self.containers.append(container_name)62        # When uploading an archived file with the bulk operation, the response63        # does not contain 'content-length' header. This is the special case,64        # therefore the existence of response headers is checked without65        # custom matcher.66        self.assertIn('transfer-encoding', resp)67        self.assertIn('content-type', resp)68        self.assertIn('x-trans-id', resp)69        self.assertIn('date', resp)70        # Check only the format of common headers with custom matcher71        self.assertThat(resp, custom_matchers.AreAllWellFormatted())72        param = {'format': 'json'}73        resp, body = self.account_client.list_account_containers(param)74        self.assertHeaders(resp, 'Account', 'GET')75        self.assertIn(container_name, [b['name'] for b in body])76        param = {'format': 'json'}77        resp, contents_list = self.container_client.list_container_contents(78            container_name, param)79        self.assertHeaders(resp, 'Container', 'GET')80        self.assertIn(object_name, [c['name'] for c in contents_list])81    @test.idempotent_id('c075e682-0d2a-43b2-808d-4116200d736d')82    @test.requires_ext(extension='bulk', service='object')83    def test_bulk_delete(self):84        # Test bulk operation of deleting multiple files85        filepath, container_name, object_name = self._create_archive()86        self._upload_archive(filepath)87        data = '%s/%s\n%s' % (container_name, object_name, container_name)88        params = {'bulk-delete': ''}89        resp, body = self.account_client.delete_account(data=data,90                                                        params=params)91        # When deleting multiple files using the bulk operation, the response92        # does not contain 'content-length' header. This is the special case,93        # therefore the existence of response headers is checked without94        # custom matcher.95        self.assertIn('transfer-encoding', resp)96        self.assertIn('content-type', resp)97        self.assertIn('x-trans-id', resp)98        self.assertIn('date', resp)99        # Check only the format of common headers with custom matcher100        self.assertThat(resp, custom_matchers.AreAllWellFormatted())101        # Check if uploaded contents are completely deleted102        self._check_contents_deleted(container_name)103    @test.idempotent_id('dbea2bcb-efbb-4674-ac8a-a5a0e33d1d79')104    @test.requires_ext(extension='bulk', service='object')105    def test_bulk_delete_by_POST(self):106        # Test bulk operation of deleting multiple files107        filepath, container_name, object_name = self._create_archive()108        self._upload_archive(filepath)109        data = '%s/%s\n%s' % (container_name, object_name, container_name)110        params = {'bulk-delete': ''}111        resp, body = self.account_client.create_account_metadata(112            {}, data=data, params=params)113        # When deleting multiple files using the bulk operation, the response114        # does not contain 'content-length' header. This is the special case,115        # therefore the existence of response headers is checked without116        # custom matcher.117        self.assertIn('transfer-encoding', resp)118        self.assertIn('content-type', resp)119        self.assertIn('x-trans-id', resp)120        self.assertIn('date', resp)121        # Check only the format of common headers with custom matcher122        self.assertThat(resp, custom_matchers.AreAllWellFormatted())...

mongo_connections_test.py

Source:mongo_connections_test.py

1"""Tests for ``upload_rest_api.app`` module."""2import pymongo3import mock4def _upload_archive(client, auth):5    """Upload 1000_files.tar.gz archive.6    :returns: HTTP response7    """8    with open("tests/data/1000_files.tar.gz", "rb") as archive:9        response = client.post(10            "/v1/archives/test_project",11            input_stream=archive,12            headers=auth13        )14        assert response.status_code == 20215    return response16def test_upload_archive(app, test_auth, background_job_runner):17    """Test mongo connections for archive upload."""18    client = app.test_client()19    with mock.patch(20        "pymongo.MongoClient",21        return_value=pymongo.MongoClient()22    ) as connect:23        response = _upload_archive(client, test_auth)24        background_job_runner(client, "upload", response)25        assert connect.call_count < 1026def test_get_files(app, test_auth, background_job_runner):27    """Test mongo connections of a GET request to project root."""28    client = app.test_client()29    response = _upload_archive(client, test_auth)30    background_job_runner(client, "upload", response)31    with mock.patch(32        "pymongo.MongoClient",33        return_value=pymongo.MongoClient()34    ) as connect:35        # GET whole project36        client.get("/v1/files/test_project/?all=true", headers=test_auth)37        assert connect.call_count == 238def test_delete_files(app, test_auth, requests_mock, background_job_runner):39    """Test mongo connections of project deletion."""40    client = app.test_client()41    response = _upload_archive(client, test_auth)42    background_job_runner(client, "upload", response)43    # Mock Metax44    response = {45        "next": None,46        "results": [47            {48                "id": "foo",49                "identifier": "foo",50                "file_path": "test/%s.txt" % i,51                "file_storage": {52                    "identifier": "urn:nbn:fi:att:file-storage-pas"53                }54            } for i in range(1000)55        ]56    }57    requests_mock.get("https://metax.localdomain/rest/v2/files?limit=10000&"58                      "project_identifier=test_project",59                      json=response)60    with mock.patch(61        "pymongo.MongoClient",62        return_value=pymongo.MongoClient()63    ) as connect:64        # DELETE the whole project65        response = client.delete(66            "/v1/files/test_project",67            headers=test_auth68        )69        background_job_runner(client, "files", response)70        assert connect.call_count < 1071def test_post_metadata(app, test_auth, mock_redis, background_job_runner):72    """Test posting file metadata to Metax."""73    client = app.test_client()74    response = _upload_archive(client, test_auth)75    background_job_runner(client, "upload", response)76    with mock.patch(77        "pymongo.MongoClient",78        return_value=pymongo.MongoClient()79    ) as connect:80        response = client.post("/v1/metadata/test_project/test/",81                               headers=test_auth)82        assert response.status_code == 20283        assert connect.call_count > 084        assert connect.call_count < 1085    # Remove the locks86    mock_redis.flushall()87def test_delete_metadata(app, test_auth, requests_mock, background_job_runner):88    """Test mongo connections of metadata deletion."""89    client = app.test_client()90    response = _upload_archive(client, test_auth)91    background_job_runner(client, "upload", response)92    # Mock Metax93    response = {94        "next": None,95        "results": [96            {97                "id": "foo",98                "identifier": "foo",99                "file_path": "/test/%i.txt" % i,100                "file_storage": {101                    "identifier": "urn:nbn:fi:att:file-storage-pas"102                }103            } for i in range(1000)104        ]...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.