How to use _copy method in autotest

Best Python code snippet using autotest_python

test_utils.py

Source:test_utils.py

1# Copyright 2015 Hitachi Data Systems inc.2# All Rights Reserved.3#4#    Licensed under the Apache License, Version 2.0 (the "License"); you may5#    not use this file except in compliance with the License. You may obtain6#    a copy of the License at7#8#         http://www.apache.org/licenses/LICENSE-2.09#10#    Unless required by applicable law or agreed to in writing, software11#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT12#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the13#    License for the specific language governing permissions and limitations14#    under the License.15import os16import mock17from manila.data import utils as data_utils18from manila import test19from manila import utils20class CopyClassTestCase(test.TestCase):21    def setUp(self):22        super(CopyClassTestCase, self).setUp()23        src = '/path/fake/src'24        dest = '/path/fake/dst'25        ignore_list = ['item']26        self._copy = data_utils.Copy(src, dest, ignore_list)27        self._copy.total_size = 1000028        self._copy.current_size = 10029        self._copy.current_copy = {'file_path': '/fake/path', 'size': 100}30        self.mock_log = self.mock_object(data_utils, 'LOG')31    def test_get_progress(self):32        expected = {'total_progress': 1,33                    'current_file_path': '/fake/path',34                    'current_file_progress': 100}35        # mocks36        self.mock_object(utils, 'execute',37                         mock.Mock(return_value=("100", "")))38        # run39        out = self._copy.get_progress()40        # asserts41        self.assertEqual(expected, out)42        utils.execute.assert_called_once_with("stat", "-c", "%s", "/fake/path",43                                              run_as_root=True)44    def test_get_progress_current_copy_none(self):45        self._copy.current_copy = None46        expected = {'total_progress': 100}47        # run48        out = self._copy.get_progress()49        # asserts50        self.assertEqual(expected, out)51    def test_get_progress_exception(self):52        expected = {'total_progress': 1,53                    'current_file_path': '/fake/path',54                    'current_file_progress': 0}55        # mocks56        self.mock_object(57            utils, 'execute',58            mock.Mock(side_effect=utils.processutils.ProcessExecutionError()))59        # run60        out = self._copy.get_progress()61        # asserts62        self.assertEqual(expected, out)63        utils.execute.assert_called_once_with("stat", "-c", "%s", "/fake/path",64                                              run_as_root=True)65    def test_cancel(self):66        self._copy.cancelled = False67        # run68        self._copy.cancel()69        # asserts70        self.assertEqual(self._copy.cancelled, True)71        # reset72        self._copy.cancelled = False73    def test_get_total_size(self):74        self._copy.total_size = 075        values = [("folder1/\nitem/\nfile1\nitem", ""),76                  ("", ""),77                  ("10000", "")]78        def get_output(*args, **kwargs):79            return values.pop(0)80        # mocks81        self.mock_object(utils, 'execute', mock.Mock(82            side_effect=get_output))83        # run84        self._copy.get_total_size(self._copy.src)85        # asserts86        self.assertEqual(self._copy.total_size, 10000)87        utils.execute.assert_has_calls([88            mock.call("ls", "-pA1", "--group-directories-first",89                      self._copy.src, run_as_root=True),90            mock.call("ls", "-pA1", "--group-directories-first",91                      os.path.join(self._copy.src, "folder1/"),92                      run_as_root=True),93            mock.call("stat", "-c", "%s",94                      os.path.join(self._copy.src, "file1"), run_as_root=True)95        ])96    def test_get_total_size_cancelled_1(self):97        self._copy.total_size = 098        self._copy.cancelled = True99        # run100        self._copy.get_total_size(self._copy.src)101        # asserts102        self.assertEqual(self._copy.total_size, 0)103        # reset104        self._copy.total_size = 10000105        self._copy.cancelled = False106    def test_get_total_size_cancelled_2(self):107        self._copy.total_size = 0108        def ls_output(*args, **kwargs):109            self._copy.cancelled = True110            return "folder1/", ""111        # mocks112        self.mock_object(utils, 'execute', mock.Mock(113            side_effect=ls_output))114        # run115        self._copy.get_total_size(self._copy.src)116        # asserts117        self.assertEqual(self._copy.total_size, 0)118        utils.execute.assert_called_once_with(119            "ls", "-pA1", "--group-directories-first", self._copy.src,120            run_as_root=True)121        # reset122        self._copy.total_size = 10000123        self._copy.cancelled = False124    def test_copy_data(self):125        values = [("folder1/\nitem/\nfile1\nitem", ""),126                  "",127                  ("", ""),128                  ("10000", ""),129                  ""]130        def get_output(*args, **kwargs):131            return values.pop(0)132        # mocks133        self.mock_object(utils, 'execute', mock.Mock(134            side_effect=get_output))135        self.mock_object(self._copy, 'get_progress')136        # run137        self._copy.copy_data(self._copy.src)138        # asserts139        self._copy.get_progress.assert_called_once_with()140        utils.execute.assert_has_calls([141            mock.call("ls", "-pA1", "--group-directories-first",142                      self._copy.src, run_as_root=True),143            mock.call("mkdir", "-p", os.path.join(self._copy.dest, "folder1/"),144                      run_as_root=True),145            mock.call("ls", "-pA1", "--group-directories-first",146                      os.path.join(self._copy.src, "folder1/"),147                      run_as_root=True),148            mock.call("stat", "-c", "%s",149                      os.path.join(self._copy.src, "file1"), run_as_root=True),150            mock.call("cp", "-P", "--preserve=all",151                      os.path.join(self._copy.src, "file1"),152                      os.path.join(self._copy.dest, "file1"), run_as_root=True)153        ])154    def test_copy_data_cancelled_1(self):155        self._copy.cancelled = True156        # run157        self._copy.copy_data(self._copy.src)158        # reset159        self._copy.cancelled = False160    def test_copy_data_cancelled_2(self):161        def ls_output(*args, **kwargs):162            self._copy.cancelled = True163            return "folder1/", ""164        # mocks165        self.mock_object(utils, 'execute', mock.Mock(166            side_effect=ls_output))167        # run168        self._copy.copy_data(self._copy.src)169        # asserts170        utils.execute.assert_called_once_with(171            "ls", "-pA1", "--group-directories-first", self._copy.src,172            run_as_root=True)173        # reset174        self._copy.cancelled = False175    def test_copy_stats(self):176        values = [("folder1/\nitem/\nfile1\nitem", ""),177                  ("", ""),178                  "",179                  "",180                  "",181                  "",182                  "",183                  ""]184        def get_output(*args, **kwargs):185            return values.pop(0)186        # mocks187        self.mock_object(utils, 'execute', mock.Mock(188            side_effect=get_output))189        # run190        self._copy.copy_stats(self._copy.src)191        # asserts192        utils.execute.assert_has_calls([193            mock.call("ls", "-pA1", "--group-directories-first",194                      self._copy.src, run_as_root=True),195            mock.call("ls", "-pA1", "--group-directories-first",196                      os.path.join(self._copy.src, "folder1/"),197                      run_as_root=True),198            mock.call(199                "chmod",200                "--reference=%s" % os.path.join(self._copy.src, "folder1/"),201                os.path.join(self._copy.dest, "folder1/"),202                run_as_root=True),203            mock.call(204                "touch",205                "--reference=%s" % os.path.join(self._copy.src, "folder1/"),206                os.path.join(self._copy.dest, "folder1/"),207                run_as_root=True),208            mock.call(209                "chown",210                "--reference=%s" % os.path.join(self._copy.src, "folder1/"),211                os.path.join(self._copy.dest, "folder1/"),212                run_as_root=True),213        ])214    def test_copy_stats_cancelled_1(self):215        self._copy.cancelled = True216        # run217        self._copy.copy_stats(self._copy.src)218        # reset219        self._copy.cancelled = False220    def test_copy_stats_cancelled_2(self):221        def ls_output(*args, **kwargs):222            self._copy.cancelled = True223            return "folder1/", ""224        # mocks225        self.mock_object(utils, 'execute', mock.Mock(226            side_effect=ls_output))227        # run228        self._copy.copy_stats(self._copy.src)229        # asserts230        utils.execute.assert_called_once_with(231            "ls", "-pA1", "--group-directories-first", self._copy.src,232            run_as_root=True)233        # reset234        self._copy.cancelled = False235    def test_run(self):236        # mocks237        self.mock_object(self._copy, 'get_total_size')238        self.mock_object(self._copy, 'copy_data')239        self.mock_object(self._copy, 'copy_stats')240        self.mock_object(self._copy, 'get_progress')241        # run242        self._copy.run()243        # asserts244        self.assertTrue(data_utils.LOG.info.called)245        self._copy.get_total_size.assert_called_once_with(self._copy.src)246        self._copy.copy_data.assert_called_once_with(self._copy.src)247        self._copy.copy_stats.assert_called_once_with(self._copy.src)...

test_ddo_checker_shacl.py

Source:test_ddo_checker_shacl.py

1#2# Copyright 2021 Ocean Protocol Foundation3# SPDX-License-Identifier: Apache-2.04#5import copy6import json7from pathlib import Path8import pkg_resources9from pyshacl import validate10import pytest11import rdflib12from aquarius.ddo_checker.shacl_checker import (13    validate_dict,14    parse_report_to_errors,15    CURRENT_VERSION,16)17from tests.ddos.ddo_sample1_v4 import json_dict18from tests.ddos.ddo_sample_algorithm_v4 import algorithm_ddo_sample19def test_sample_schema():20    path = "sample_schemas/remote_v4.ttl"21    schema_file = Path(pkg_resources.resource_filename("tests", path)).read_text()22    data = json.dumps(23        {24            "@context": {"@vocab": "http://schema.org/"},25            "@type": "DDO",26            "id": "123",27            "version": CURRENT_VERSION,28        }29    )30    dataGraph = rdflib.Graph().parse(data=data, format="json-ld")31    conforms, results_graph, _ = validate(dataGraph, shacl_graph=schema_file)32    assert not conforms33    errors = parse_report_to_errors(results_graph)34    assert "id" in errors35    assert errors["id"] == "Value does not match pattern '^did\:op\:(.*)$'"36    data = json.dumps(37        {38            "@context": {"@vocab": "http://schema.org/"},39            "@type": "DDO",40            "version": CURRENT_VERSION,41        }42    )43    dataGraph = rdflib.Graph().parse(data=data, format="json-ld")44    conforms, results_graph, _ = validate(dataGraph, shacl_graph=schema_file)45    assert not conforms46    errors = parse_report_to_errors(results_graph)47    assert "id" in errors48    assert "Less than 1 value on schema" in errors["id"]49    data = json.dumps(50        {51            "@context": {"@vocab": "http://schema.org/"},52            "@type": "DDO",53            "id": "did:op:123",54            "version": CURRENT_VERSION,55        }56    )57    dataGraph = rdflib.Graph().parse(data=data, format="json-ld")58    conforms, results_graph, results_text = validate(dataGraph, shacl_graph=schema_file)59    assert conforms60    errors = parse_report_to_errors(results_graph)61    assert not errors62def test_remote_ddo_passes():63    valid, _ = validate_dict(json_dict, json_dict["chainId"], json_dict["nftAddress"])64    assert valid65    valid, _ = validate_dict(66        algorithm_ddo_sample,67        algorithm_ddo_sample["chainId"],68        algorithm_ddo_sample["nftAddress"],69    )70    assert valid71def test_remote_ddo_fails():72    _copy = copy.deepcopy(json_dict)73    _copy.pop("@context")74    valid, errors = validate_dict(_copy, "", "")75    assert not valid76    assert errors["@context"] == "Context is missing or invalid."77    _copy = copy.deepcopy(json_dict)78    valid, errors = validate_dict(_copy, "", "")79    assert not valid80    assert errors["chainId"] == "chainId is missing or invalid."81    assert errors["nftAddress"] == "nftAddress is missing or invalid."82    _copy = copy.deepcopy(json_dict)83    valid, errors = validate_dict(_copy, 1234, json_dict["nftAddress"])84    assert not valid85    assert errors["id"] == "did is not valid for chain Id and nft address"86    _copy = copy.deepcopy(json_dict)87    valid, errors = validate_dict(_copy, json_dict["chainId"], "0xabcd")88    assert not valid89    assert errors["id"] == "did is not valid for chain Id and nft address"90    for required_prop in ["id", "version"]:91        _copy = copy.deepcopy(json_dict)92        _copy.pop(required_prop)93        valid, errors = validate_dict(94            _copy, json_dict["chainId"], json_dict["nftAddress"]95        )96        assert not valid97        assert required_prop in errors98    _copy = copy.deepcopy(json_dict)99    _copy["version"] = "something not semver"100    with pytest.raises(AssertionError):101        validate_dict(_copy, json_dict["chainId"], json_dict["nftAddress"])102    # services invalid103    _copy = copy.deepcopy(json_dict)104    _copy["services"][0].pop("files")105    valid, errors = validate_dict(_copy, json_dict["chainId"], json_dict["nftAddress"])106    assert not valid107    assert "services" in errors108    for required_prop in ["type", "datatokenAddress", "serviceEndpoint", "timeout"]:109        _copy = copy.deepcopy(json_dict)110        _copy["services"][0].pop(required_prop)111        valid, errors = validate_dict(112            _copy, json_dict["chainId"], json_dict["nftAddress"]113        )114        assert not valid115        assert "services" in errors116    _copy = copy.deepcopy(json_dict)117    _copy["services"][0]["timeout"] = "not an integer"118    valid, errors = validate_dict(_copy, json_dict["chainId"], json_dict["nftAddress"])119    assert not valid120    assert "services" in errors121def test_remote_ddo_failures_limits():122    # simple maxLength check (metadata name)123    _copy = copy.deepcopy(json_dict)124    _copy["metadata"]["name"] = "a" * 513125    valid, errors = validate_dict(_copy, json_dict["chainId"], json_dict["nftAddress"])126    assert not valid127    assert "metadata" in errors128    # too many tags129    _copy = copy.deepcopy(json_dict)130    _copy["metadata"]["tags"] = [str(x) for x in range(0, 65)]131    valid, errors = validate_dict(_copy, json_dict["chainId"], json_dict["nftAddress"])132    assert not valid133    assert "metadata" in errors134    # algorithm container checksum is the wrong length135    _copy = copy.deepcopy(algorithm_ddo_sample)136    _copy["metadata"]["algorithm"]["container"]["checksum"] = ("sha2:",)137    valid, errors = validate_dict(138        _copy,139        algorithm_ddo_sample["chainId"],140        algorithm_ddo_sample["nftAddress"],141    )142    assert not valid143    assert "metadata" in errors144def test_remote_ddo_metadata_fails():145    for required_prop in ["description", "name", "type", "author", "license"]:146        _copy = copy.deepcopy(json_dict)147        _copy["metadata"].pop(required_prop)148        valid, _ = validate_dict(_copy, json_dict["chainId"], json_dict["nftAddress"])149        assert not valid150    _copy = copy.deepcopy(json_dict)151    _copy["metadata"]["created"] = "not iso format"152    valid, errors = validate_dict(_copy, json_dict["chainId"], json_dict["nftAddress"])153    assert not valid154    assert errors["metadata"] == "created is not in iso format."155    _copy = copy.deepcopy(json_dict)156    _copy["metadata"]["links"] = 14  # not a link or string list157    valid, _ = validate_dict(_copy, json_dict["chainId"], json_dict["nftAddress"])158    assert not valid159    _copy = copy.deepcopy(json_dict)160    _copy["metadata"]["tags"] = 14  # not a string or string list161    valid, _ = validate_dict(_copy, json_dict["chainId"], json_dict["nftAddress"])162    assert not valid163    _copy = copy.deepcopy(algorithm_ddo_sample)164    _copy["metadata"]["algorithm"]["container"] = None165    valid, errors = validate_dict(_copy, json_dict["chainId"], json_dict["nftAddress"])166    assert not valid167def test_remote_ddo_without_service():168    _copy = copy.deepcopy(json_dict)169    _ = _copy.pop("services")170    assert not _copy.get("services")171    assert _copy["version"] == CURRENT_VERSION172    valid, _ = validate_dict(173        _copy,174        _copy["chainId"],175        _copy["nftAddress"],176    )...

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.