Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use artifacts_base_dir method in yandex-tank

Best Python code snippet using yandex-tank

artifacts_test.py

Source:artifacts_test.py

1# Copyright 2019 Google Inc. All rights reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#    http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14import hashlib15import logging16import os17import shutil18import sys19import tempfile20import unittest21from typ import artifacts22from typ.fakes.host_fake import FakeHost23class ArtifactsArtifactCreationTests(unittest.TestCase):24  def test_create_artifact_writes_to_disk_iteration_0_no_test_dir(self):25    host = FakeHost()26    output_dir = '%stmp' % host.sep27    ar = artifacts.Artifacts(output_dir, host)28    file_rel_path = host.join('stdout', 'test.jpg')29    ar.CreateArtifact('artifact_name', file_rel_path, b'contents')30    self.assertEqual(31        host.read_binary_file(32            host.join(output_dir, 'stdout', 'test.jpg')),33        b'contents')34  def test_create_artifact_writes_to_disk_iteration_1_no_test_dir(self):35    host = FakeHost()36    output_dir = '%stmp' % host.sep37    ar = artifacts.Artifacts(38      output_dir, host, iteration=1)39    file_rel_path = host.join('stdout', 'test.jpg')40    ar.CreateArtifact('artifact_name', file_rel_path, b'contents')41    self.assertEqual(42        host.read_binary_file(43            host.join(output_dir, 'retry_1', 'stdout', 'test.jpg')),44        b'contents')45  def test_create_artifact_writes_to_disk_iteration_1_test_dir(self):46    host = FakeHost()47    output_dir = '%stmp' % host.sep48    ar = artifacts.Artifacts(49      output_dir, host, iteration=1, artifacts_base_dir='a.b.c')50    file_rel_path = host.join('stdout', 'test.jpg')51    ar.CreateArtifact('artifact_name', file_rel_path, b'contents')52    self.assertEqual(53        host.read_binary_file(54            host.join(output_dir, 'a.b.c', 'retry_1', 'stdout', 'test.jpg')),55        b'contents')56  def test_overwriting_artifact_raises_value_error(self):57    host = FakeHost()58    output_dir = '%stmp' % host.sep59    ar = artifacts.Artifacts(60      output_dir, host, iteration=0, artifacts_base_dir='retry_1')61    file_rel_path = host.join('stdout', 'test.jpg')62    ar.CreateArtifact('artifact_name', file_rel_path, b'contents')63    ar1 = artifacts.Artifacts(64      output_dir, host, iteration=1)65    with self.assertRaises(ValueError) as ve:66        ar1.CreateArtifact('artifact_name', file_rel_path, b'overwritten contents')67    self.assertIn('already exists', str(ve.exception))68  def test_force_overwriting_artifact_does_not_raise_error(self):69    host = FakeHost()70    output_dir = '%stmp' % host.sep71    ar = artifacts.Artifacts(72      output_dir, host, iteration=0, artifacts_base_dir='a.b.c', intial_results_base_dir=True)73    file_rel_path = host.join('stdout', 'test.txt')74    ar.CreateArtifact('artifact_name', file_rel_path, 'contents',75                      write_as_text=True)76    self.assertEqual(77        host.read_text_file(78            host.join(output_dir, 'a.b.c', 'initial', 'stdout', 'test.txt')),79        'contents')80    ar.CreateArtifact('artifact_name', file_rel_path, 'overwritten contents',81                      force_overwrite=True, write_as_text=True)82    self.assertEqual(83        host.read_text_file(84            host.join(output_dir, 'a.b.c', 'initial', 'stdout', 'test.txt')),85        'overwritten contents')86  def test_create_artifact_writes_to_disk_initial_results_dir(self):87    host = FakeHost()88    output_dir = '%stmp' % host.sep89    ar = artifacts.Artifacts(90      output_dir, host, iteration=0, artifacts_base_dir='a.b.c', intial_results_base_dir=True)91    file_rel_path = host.join('stdout', 'test.jpg')92    ar.CreateArtifact('artifact_name', file_rel_path, b'contents')93    self.assertEqual(94        host.read_binary_file(host.join(output_dir, 'a.b.c', 'initial', 'stdout', 'test.jpg')),95        b'contents')96  def test_file_manager_writes_file(self):97    host = FakeHost()98    output_dir = '%stmp' % host.sep99    ar = artifacts.Artifacts(output_dir, host, iteration=0)100    file_path = host.join('failures', 'stderr.txt')101    ar.CreateArtifact('artifact_name', file_path, 'exception raised',102                      write_as_text=True)103    self.assertEqual(104        host.read_text_file(file_path), 'exception raised')105  def test_duplicate_artifact_raises_error_when_added_to_list(self):106    host = FakeHost()107    output_dir = '%stmp' % host.sep108    ar = artifacts.Artifacts(output_dir, host, iteration=0)109    ar.AddArtifact('artifact_name', 'foo.txt')110    with self.assertRaises(ValueError) as ve:111        ar.AddArtifact('artifact_name', 'foo.txt')112    self.assertIn('already exists', str(ve.exception))113  def test_dont_raise_value_error_for_dupl_in_add_artifacts(self):114    host = FakeHost()115    output_dir = '%stmp' % host.sep116    ar = artifacts.Artifacts(output_dir, host, iteration=0)117    ar.AddArtifact('artifact_name', 'foo.txt')118    ar.AddArtifact('artifact_name', 'foo.txt',119                    raise_exception_for_duplicates=False)120    self.assertEqual(ar.artifacts['artifact_name'], ['foo.txt'])121  def test_windows_path_limit_workaround(self):122    host = FakeHost()123    host.platform = 'win32'124    output_dir = '%stmp' % host.sep125    artifacts_base_dir = 'a' * artifacts.WINDOWS_MAX_PATH126    ar = artifacts.Artifacts(output_dir, host, iteration=0,127                             artifacts_base_dir=artifacts_base_dir)128    file_path = host.join('failures', 'stderr.txt')129    ar.CreateArtifact('artifact_name', file_path, 'exception raised',130                      write_as_text=True)131    m = hashlib.sha1()132    m.update(artifacts_base_dir.encode('utf-8'))133    expected_dir = m.hexdigest()134    expected_path = host.join(output_dir, expected_dir, file_path)135    self.assertEqual(host.read_text_file(expected_path), 'exception raised')136  def test_windows_path_limit_workaround_too_long(self):137    host = FakeHost()138    host.platform = 'win32'139    output_dir = '%stmp' % host.sep140    ar = artifacts.Artifacts(output_dir, host, iteration=0)141    file_rel_path = 'a' * (artifacts.WINDOWS_MAX_PATH)142    if host.is_python3:143      with self.assertLogs(logging.getLogger(), logging.ERROR) as output:144        ar.CreateArtifact('artifact_name', file_rel_path, b'contents')145      for log_line in output.output:146        if 'exceeds Windows MAX_PATH' in log_line:147          break148      else:149        self.fail('Did not get expected log line')150    else:151      ar.CreateArtifact('artifact_name', file_rel_path, b'contents')152    self.assertEqual(ar.artifacts, {})153  def test_mac_path_limit_workaround(self):154    host = FakeHost()155    host.platform = 'darwin'156    output_dir = '%stmp' % host.sep157    long_artifact_piece = 'a' * (artifacts.MAC_MAX_FILE_NAME + 1)158    long_relative_piece = 'b' * (artifacts.MAC_MAX_FILE_NAME + 1)159    artifacts_base_dir = host.join(long_artifact_piece, 'short')160    ar = artifacts.Artifacts(output_dir, host, iteration=0,161                             artifacts_base_dir=artifacts_base_dir)162    file_rel_path = host.join(long_relative_piece, 'output.txt')163    ar.CreateArtifact('artifact_name', file_rel_path, b'content')164    m = hashlib.sha1()165    m.update(long_artifact_piece.encode('utf-8'))166    artifact_hash = m.hexdigest()167    m = hashlib.sha1()168    m.update(long_relative_piece.encode('utf-8'))169    relative_hash = m.hexdigest()170    expected_path = host.join(171        output_dir, artifact_hash, 'short', relative_hash, 'output.txt')172    self.assertEqual(host.read_binary_file(expected_path), b'content')173class ArtifactsLinkCreationTests(unittest.TestCase):174  def test_create_link(self):175    ar = artifacts.Artifacts('', FakeHost())176    ar.CreateLink('link', 'https://testsite.com')177    self.assertEqual(ar.artifacts, {'link': ['https://testsite.com']})178  def test_create_link_invalid_url(self):179    ar = artifacts.Artifacts('', FakeHost())180    with self.assertRaises(ValueError):181      ar.CreateLink('link', 'https:/malformedurl.com')182  def test_create_link_non_https(self):183    ar = artifacts.Artifacts('', FakeHost())184    with self.assertRaises(ValueError):185      ar.CreateLink('link', 'http://testsite.com')186  def test_create_link_newlines(self):187    ar = artifacts.Artifacts('', FakeHost())188    with self.assertRaises(ValueError):...

build.py

Source:build.py

1import os2import subprocess3import sys4from argparse import ArgumentParser5from os.path import abspath6from shutil import rmtree7from typing import List8import jinja29import pkg_resources10from ophiuchus.cli.subcommands import EntryPointBuilderSubcommand11from ophiuchus.framework import GlobalConfig12from ophiuchus.framework import Handler13from ophiuchus.utils import load_entry_points14from pip import main as pip15def python_version():16    return f"{sys.version_info.major}.{sys.version_info.minor}"17class Build(EntryPointBuilderSubcommand):18    description = "Build website lambdas"19    def __init__(self, parser: ArgumentParser):20        super().__init__(parser)21        parser.add_argument(22            "--artifacts-base-dir",23            default="./build",24            type=abspath,25            help="Build/install directory for artifacts",26        )27        parser.add_argument(28            "--python-version",29            default=python_version(),30            type=str,31            help="Python version to build/package for. (Default is the current "32            "version: %(default)s)",33        )34        parser.add_argument(35            "--requirements-file",36            default="./requirements.txt",37            type=abspath,38            help="Path to application requirements.txt file",39        )40    def __call__(41        self,42        site_groups: List[str],43        artifacts_base_dir: str,44        additional_endpoints: List[List[str]] = [],45        python_version: str = python_version(),46        requirements_file: str = "./requirements.txt",47        *args,48        **kwargs,49    ):50        config = GlobalConfig(  # noqa: F84151            endpoints=dict(additional_endpoints),52        )53        self.log.info(f"Cleaning old artifacts dir: {artifacts_base_dir}")54        rmtree(artifacts_base_dir)55        os.makedirs(artifacts_base_dir, exist_ok=True)56        for site_group in site_groups:57            self.build_site_group(58                site_group=site_group,59                artifacts_base_dir=artifacts_base_dir,60                python_version=python_version,61                requirements_file=requirements_file,62            )63    def build_site_group(64        self,65        site_group: str,66        artifacts_base_dir: str,67        python_version: str = python_version(),68        requirements_file: str = "./requirements.txt",69    ):70        self.log.info(f"Building {site_group}")71        site_group_artifact_dir = os.path.join(artifacts_base_dir, site_group,)72        self.log.debug(f"Using artifact path: {site_group_artifact_dir}")73        site_group_lambda_dir = os.path.join(74            site_group_artifact_dir, "lambdas",75        )76        self.log.debug(f"Using lambda path: {site_group_lambda_dir}")77        site_group_packages_dir = os.path.join(78            site_group_artifact_dir,79            "python",80            "lib",81            f"python{python_version}",82            "site-packages",83        )84        self.log.debug(85            f"Using package install path: {site_group_packages_dir}",86        )87        self.log.debug("Creating site group artifact directory")88        os.makedirs(site_group_artifact_dir, exist_ok=True)89        self.install_package(90            site_group_packages_dir=site_group_packages_dir,91            requirements_file=requirements_file,92        )93        self.build_lambdas(94            site_group=site_group,95            site_group_lambda_dir=site_group_lambda_dir,96            site_group_packages_dir=site_group_packages_dir,97        )98    def install_package(99        self,100        site_group_packages_dir: str,101        requirements_file: str = "./requirements.txt",102    ):103        self.log.debug("Creating site group packages directory")104        os.makedirs(site_group_packages_dir, exist_ok=True)105        self.log.info(106            f"Installing package from requirements file {requirements_file} "107            f"into {site_group_packages_dir}",108        )109        subprocess.run(110            [111                "pip",112                "install",113                "--target",114                site_group_packages_dir,115                "-r",116                requirements_file,117            ],118            stdout=subprocess.PIPE,119            stderr=subprocess.PIPE,120        )121    def build_lambdas(122        self,123        site_group: str,124        site_group_lambda_dir: str,125        site_group_packages_dir: str,126    ):127        self.log.debug("Creating site group lambda directory")128        os.makedirs(site_group_lambda_dir, exist_ok=True)129        working_set = pkg_resources.WorkingSet(130            entries=[site_group_packages_dir],131        )132        env = jinja2.Environment(133            loader=jinja2.PackageLoader("ophiuchus", "templates"),134        )135        template = env.get_template("lambdas/handler.py.template")136        for name, ep in load_entry_points(137            site_group, Handler, working_set,138        ).items():139            file_name = os.path.join(site_group_lambda_dir, f"{name}.py")140            with open(file_name, "w") as f:141                f.write(142                    template.render(module=ep.__module__, name=ep.__name__),...

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.