KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use mkstemp method in yandex-tank

Best Python code snippet using yandex-tank

test_file_helpers.py

Source:test_file_helpers.py

1# -*- coding: utf-8 -*-2from __future__ import absolute_import, division, print_function, unicode_literals3__license__ = "GNU Affero General Public License http://www.gnu.org/licenses/agpl.html"4__copyright__ = "Copyright (C) 2015 The OctoPrint Project - Released under terms of the AGPLv3 License"5import io6import os7import sys8import unittest9import ddt10import mock11from past.builtins import unicode12import octoprint.util13class BomAwareOpenTest(unittest.TestCase):14    """15    Tests for :func:`octoprint.util.bom_aware_open`.16    """17    def setUp(self):18        self.filename_utf8_with_bom = os.path.join(19            os.path.abspath(os.path.dirname(__file__)), "_files", "utf8_with_bom.txt"20        )21        self.filename_utf8_without_bom = os.path.join(22            os.path.abspath(os.path.dirname(__file__)), "_files", "utf8_without_bom.txt"23        )24    def test_bom_aware_open_with_bom(self):25        """Tests that the contents of a UTF8 file with BOM are loaded correctly (without the BOM)."""26        # test27        with octoprint.util.bom_aware_open(28            self.filename_utf8_with_bom, encoding="utf-8"29        ) as f:30            contents = f.readlines()31        # assert32        self.assertEqual(len(contents), 3)33        self.assertTrue(contents[0].startswith("#"))34    def test_bom_aware_open_without_bom(self):35        """Tests that the contents of a UTF8 file without BOM are loaded correctly."""36        # test37        with octoprint.util.bom_aware_open(38            self.filename_utf8_without_bom, encoding="utf-8"39        ) as f:40            contents = f.readlines()41        # assert42        self.assertEqual(len(contents), 3)43        self.assertTrue(contents[0].startswith("#"))44    def test_bom_aware_open_ascii(self):45        """Tests that the contents of a UTF8 file loaded as ASCII are replaced correctly if "replace" is specified on errors."""46        # test47        with octoprint.util.bom_aware_open(48            self.filename_utf8_with_bom, errors="replace"49        ) as f:50            contents = f.readlines()51        # assert52        self.assertEqual(len(contents), 3)53        self.assertTrue(contents[0].startswith("\ufffd" * 3 + "#"))54        self.assertTrue(contents[2].endswith("\ufffd\ufffd" * 6))55    def test_bom_aware_open_encoding_error(self):56        """Tests that an encoding error is thrown if not suppressed when opening a UTF8 file as ASCII."""57        try:58            with octoprint.util.bom_aware_open(self.filename_utf8_without_bom) as f:59                f.readlines()60            self.fail("Expected an exception")61        except UnicodeDecodeError:62            pass63    def test_bom_aware_open_parameters_text_mode(self):64        """Tests that the parameters are propagated properly in text mode."""65        with mock.patch("io.open", wraps=io.open) as mock_open:66            with octoprint.util.bom_aware_open(67                self.filename_utf8_without_bom,68                mode="rt",69                encoding="utf-8",70                errors="ignore",71            ) as f:72                f.readlines()73        calls = [74            mock.call(self.filename_utf8_without_bom, mode="rb"),75            mock.call(76                self.filename_utf8_without_bom,77                encoding="utf-8",78                mode="rt",79                errors="ignore",80            ),81        ]82        mock_open.assert_has_calls(calls)83    def test_bom_aware_open_parameters_binary_mode(self):84        """Tests that binary mode raises an AssertionError."""85        self.assertRaises(86            AssertionError,87            octoprint.util.bom_aware_open,88            self.filename_utf8_without_bom,89            mode="rb",90            encoding="utf-8",91            errors="ignore",92        )93class TestAtomicWrite(unittest.TestCase):94    """95    Tests for :func:`octoprint.util.atomic_write`.96    """97    def setUp(self):98        pass99    @mock.patch("shutil.move")100    @mock.patch("tempfile.mkstemp")101    @mock.patch("io.open")102    @mock.patch("os.close")103    @mock.patch("os.chmod")104    @mock.patch("os.path.exists")105    def test_atomic_write(106        self, mock_exists, mock_chmod, mock_close, mock_open, mock_mkstemp, mock_move107    ):108        """Tests the regular basic "good" case."""109        # setup110        fd = 0111        path = "tempfile.tmp"112        umask = 0o026113        mock_file = mock.MagicMock()114        mock_file.name = path115        mock_mkstemp.return_value = fd, path116        mock_open.return_value = mock_file117        mock_exists.return_value = False118        # test119        with mock.patch("octoprint.util.UMASK", umask):120            with octoprint.util.atomic_write("somefile.yaml") as f:121                f.write("test")122        # assert123        mock_mkstemp.assert_called_once_with(prefix="tmp", suffix="", dir="")124        mock_close.assert_called_once_with(fd)125        mock_open.assert_called_once_with(path, mode="w+b")126        mock_file.write.assert_called_once_with("test")127        mock_file.close.assert_called_once_with()128        mock_chmod.assert_called_once_with(path, 0o644 & ~umask)129        mock_move.assert_called_once_with(path, "somefile.yaml")130    @mock.patch("shutil.move")131    @mock.patch("tempfile.mkstemp")132    @mock.patch("io.open")133    @mock.patch("os.close")134    @mock.patch("os.chmod")135    @mock.patch("os.path.exists")136    def test_atomic_write_path_aware(137        self, mock_exists, mock_chmod, mock_close, mock_open, mock_mkstemp, mock_move138    ):139        """Tests whether the tempoary file is to created in the same directory as the target file."""140        # setup141        fd = 0142        tmpdirpath = "/testpath/with/subdirectories"143        path = os.path.join(tmpdirpath, "tempfile.tmp")144        targetpath = os.path.join(tmpdirpath, "somefile.yaml")145        mock_file = mock.MagicMock()146        mock_file.name = path147        mock_mkstemp.return_value = fd, path148        mock_open.return_value = mock_file149        mock_exists.return_value = False150        # test151        with octoprint.util.atomic_write(targetpath) as f:152            f.write("test")153        # assert154        mock_mkstemp.assert_called_once_with(prefix="tmp", suffix="", dir=tmpdirpath)155        mock_open.assert_called_once_with(path, mode="w+b")156        mock_move.assert_called_once_with(path, targetpath)157    @mock.patch("shutil.move")158    @mock.patch("tempfile.mkstemp")159    @mock.patch("io.open")160    @mock.patch("os.close")161    @mock.patch("os.chmod")162    @mock.patch("os.path.exists")163    def test_atomic_write_rel_path_aware(164        self, mock_exists, mock_chmod, mock_close, mock_open, mock_mkstemp, mock_move165    ):166        """Tests whether the tempoary file is to created in the same directory as the target file. This time submitting a relative path. """167        # setup168        fd = 0169        tmpdirpath = "../test"170        path = os.path.join(tmpdirpath, "tempfile.tmp")171        targetpath = os.path.join(tmpdirpath, "somefile.yaml")172        mock_file = mock.MagicMock()173        mock_file.name = path174        mock_mkstemp.return_value = fd, path175        mock_open.return_value = mock_file176        mock_exists.return_value = False177        # test178        with octoprint.util.atomic_write(targetpath) as f:179            f.write("test")180        # assert181        mock_mkstemp.assert_called_once_with(prefix="tmp", suffix="", dir=tmpdirpath)182        mock_open.assert_called_once_with(path, mode="w+b")183        mock_move.assert_called_once_with(path, targetpath)184    @mock.patch("shutil.move")185    @mock.patch("tempfile.mkstemp")186    @mock.patch("io.open")187    @mock.patch("os.close")188    @mock.patch("os.chmod")189    @mock.patch("os.path.exists")190    def test_atomic_write_error_on_write(191        self, mock_exists, mock_chmod, mock_close, mock_open, mock_mkstemp, mock_move192    ):193        """Tests the error case where something in the wrapped code fails."""194        # setup195        fd = 0196        path = "tempfile.tmp"197        mock_file = mock.MagicMock()198        mock_file.name = path199        mock_file.write.side_effect = RuntimeError()200        mock_mkstemp.return_value = fd, path201        mock_open.return_value = mock_file202        mock_exists.return_value = False203        # test204        try:205            with octoprint.util.atomic_write("somefile.yaml") as f:206                f.write("test")207            self.fail("Expected an exception")208        except RuntimeError:209            pass210        # assert211        mock_mkstemp.assert_called_once_with(prefix="tmp", suffix="", dir="")212        mock_close.assert_called_once_with(fd)213        mock_open.assert_called_once_with(path, mode="w+b")214        mock_file.close.assert_called_once_with()215        self.assertFalse(mock_move.called)216        self.assertFalse(mock_chmod.called)217    @mock.patch("shutil.move")218    @mock.patch("tempfile.mkstemp")219    @mock.patch("io.open")220    @mock.patch("os.close")221    @mock.patch("os.chmod")222    @mock.patch("os.path.exists")223    def test_atomic_write_error_on_move(224        self, mock_exists, mock_chmod, mock_close, mock_open, mock_mkstemp, mock_move225    ):226        """Tests the error case where the final move fails."""227        # setup228        fd = 0229        path = "tempfile.tmp"230        mock_file = mock.MagicMock()231        mock_file.name = path232        mock_mkstemp.return_value = fd, path233        mock_open.return_value = mock_file234        mock_move.side_effect = RuntimeError()235        mock_exists.return_value = False236        # test237        try:238            with octoprint.util.atomic_write("somefile.yaml") as f:239                f.write("test")240            self.fail("Expected an exception")241        except RuntimeError:242            pass243        # assert244        mock_mkstemp.assert_called_once_with(prefix="tmp", suffix="", dir="")245        mock_close.assert_called_once_with(fd)246        mock_open.assert_called_once_with(path, mode="w+b")247        mock_file.close.assert_called_once_with()248        self.assertTrue(mock_move.called)249        self.assertTrue(mock_chmod.called)250    @mock.patch("shutil.move")251    @mock.patch("tempfile.mkstemp")252    @mock.patch("io.open")253    @mock.patch("os.close")254    @mock.patch("os.chmod")255    @mock.patch("os.path.exists")256    def test_atomic_write_parameters(257        self, mock_exists, mock_chmod, mock_close, mock_open, mock_mkstemp, mock_move258    ):259        """Tests that the open parameters are propagated properly."""260        # setup261        fd = 0262        path = "tempfile.tmp"263        umask = 0o026264        mock_file = mock.MagicMock()265        mock_file.name = path266        mock_mkstemp.return_value = fd, path267        mock_open.return_value = mock_file268        mock_exists.return_value = False269        # test270        with mock.patch("octoprint.util.UMASK", umask):271            with octoprint.util.atomic_write(272                "somefile.yaml", mode="w", prefix="foo", suffix="bar"273            ) as f:274                f.write("test")275        # assert276        mock_mkstemp.assert_called_once_with(prefix="foo", suffix="bar", dir="")277        mock_close.assert_called_once_with(fd)278        mock_open.assert_called_once_with(path, mode="w", encoding="utf-8")279        mock_file.close.assert_called_once_with()280        mock_chmod.assert_called_once_with(path, 0o664 & ~umask)281        mock_move.assert_called_once_with(path, "somefile.yaml")282    @mock.patch("shutil.move")283    @mock.patch("tempfile.mkstemp")284    @mock.patch("io.open")285    @mock.patch("os.close")286    @mock.patch("os.chmod")287    @mock.patch("os.path.exists")288    def test_atomic_write_custom_permissions(289        self, mock_exists, mock_chmod, mock_close, mock_open, mock_mkstemp, mock_move290    ):291        """Tests that custom permissions may be set."""292        # setup293        fd = 0294        path = "tempfile.tmp"295        mock_file = mock.MagicMock()296        mock_file.name = path297        mock_mkstemp.return_value = fd, path298        mock_open.return_value = mock_file299        mock_exists.return_value = False300        # test301        with octoprint.util.atomic_write(302            "somefile.yaml", mode="wt", permissions=0o755303        ) as f:304            f.write("test")305        # assert306        mock_mkstemp.assert_called_once_with(prefix="tmp", suffix="", dir="")307        mock_close.assert_called_once_with(fd)308        mock_open.assert_called_once_with(path, mode="wt", encoding="utf-8")309        mock_file.close.assert_called_once_with()310        mock_chmod.assert_called_once_with(path, 0o755)311        mock_move.assert_called_once_with(path, "somefile.yaml")312    @mock.patch("shutil.move")313    @mock.patch("tempfile.mkstemp")314    @mock.patch("io.open")315    @mock.patch("os.close")316    @mock.patch("os.chmod")317    @mock.patch("os.path.exists")318    @mock.patch("os.stat")319    def test_atomic_permissions_combined(320        self,321        mock_stat,322        mock_exists,323        mock_chmod,324        mock_close,325        mock_open,326        mock_mkstemp,327        mock_move,328    ):329        """Tests that the permissions of an existing file are combined with the requested permissions."""330        # setup331        fd = 0332        path = "tempfile.tmp"333        mock_file = mock.MagicMock()334        mock_file.name = path335        mock_mkstemp.return_value = fd, path336        mock_open.return_value = mock_file337        mock_exists.return_value = True338        mock_stat_result = mock.MagicMock()339        mock_stat_result.st_mode = 0o666340        mock_stat.return_value = mock_stat_result341        # test342        with octoprint.util.atomic_write(343            "somefile.yaml", mode="wt", permissions=0o755344        ) as f:345            f.write("test")346        # assert347        mock_mkstemp.assert_called_once_with(prefix="tmp", suffix="", dir="")348        mock_close.assert_called_once_with(fd)349        mock_open.assert_called_once_with(path, mode="wt", encoding="utf-8")350        mock_file.close.assert_called_once_with()351        mock_chmod.assert_called_once_with(path, 0o777)  # 0o755 | 0o666352        mock_move.assert_called_once_with(path, "somefile.yaml")353    @mock.patch("shutil.move")354    @mock.patch("tempfile.mkstemp")355    @mock.patch("io.open")356    @mock.patch("os.close")357    @mock.patch("os.chmod")358    @mock.patch("os.path.exists")359    @mock.patch("os.stat")360    def test_atomic_permissions_limited(361        self,362        mock_stat,363        mock_exists,364        mock_chmod,365        mock_close,366        mock_open,367        mock_mkstemp,368        mock_move,369    ):370        """Tests that max_permissions limit the combined file permissions."""371        # setup372        fd = 0373        path = "tempfile.tmp"374        mock_file = mock.MagicMock()375        mock_file.name = path376        mock_mkstemp.return_value = fd, path377        mock_open.return_value = mock_file378        mock_exists.return_value = True379        mock_stat_result = mock.MagicMock()380        mock_stat_result.st_mode = 0o755381        mock_stat.return_value = mock_stat_result382        # test383        with octoprint.util.atomic_write(384            "somefile.yaml", mode="wt", permissions=0o600, max_permissions=0o666385        ) as f:386            f.write("test")387        # assert388        mock_mkstemp.assert_called_once_with(prefix="tmp", suffix="", dir="")389        mock_close.assert_called_once_with(fd)390        mock_open.assert_called_once_with(path, mode="wt", encoding="utf-8")391        mock_file.close.assert_called_once_with()392        mock_chmod.assert_called_once_with(393            path, 0o644394        )  # (0o600 | 0o755) & 0o666 = 0o755 & 0o666395        mock_move.assert_called_once_with(path, "somefile.yaml")396class TempDirTest(unittest.TestCase):397    @mock.patch("shutil.rmtree")398    @mock.patch("tempfile.mkdtemp")399    def test_tempdir(self, mock_mkdtemp, mock_rmtree):400        """Tests regular "good" case."""401        # setup402        path = "/path/to/tmpdir"403        mock_mkdtemp.return_value = path404        # test405        with octoprint.util.tempdir() as td:406            self.assertEqual(td, path)407        # assert408        mock_mkdtemp.assert_called_once_with()409        mock_rmtree.assert_called_once_with(path, ignore_errors=False, onerror=None)410    @mock.patch("shutil.rmtree")411    @mock.patch("tempfile.mkdtemp")412    def test_tempdir_parameters_mkdtemp(self, mock_mkdtemp, mock_rmtree):413        """Tests that parameters for mkdtemp are properly propagated."""414        # setup415        path = "/path/to/tmpdir"416        mock_mkdtemp.return_value = path417        # test418        with octoprint.util.tempdir(prefix="prefix", suffix="suffix", dir="dir") as td:419            self.assertEqual(td, path)420        # assert421        mock_mkdtemp.assert_called_once_with(prefix="prefix", suffix="suffix", dir="dir")422        mock_rmtree.assert_called_once_with(path, ignore_errors=False, onerror=None)423    @mock.patch("shutil.rmtree")424    @mock.patch("tempfile.mkdtemp")425    def test_tempdir_parameters_rmtree(self, mock_mkdtemp, mock_rmtree):426        """Tests that parameters for rmtree are properly propagated."""427        # setup428        path = "/path/to/tmpdir"429        mock_mkdtemp.return_value = path430        onerror = mock.MagicMock()431        # test432        with octoprint.util.tempdir(ignore_errors=True, onerror=onerror) as td:433            self.assertEqual(td, path)434        # assert435        mock_mkdtemp.assert_called_once_with()436        mock_rmtree.assert_called_once_with(path, ignore_errors=True, onerror=onerror)437@ddt.ddt438class IsHiddenPathTest(unittest.TestCase):439    def setUp(self):440        import tempfile441        self.basepath = tempfile.mkdtemp()442        self.path_always_visible = os.path.join(self.basepath, "always_visible.txt")443        self.path_hidden_on_windows = os.path.join(self.basepath, "hidden_on_windows.txt")444        self.path_always_hidden = os.path.join(self.basepath, ".always_hidden.txt")445        import sys446        for attr in (447            "path_always_visible",448            "path_hidden_on_windows",449            "path_always_hidden",450        ):451            path = getattr(self, attr)452            with io.open(path, "wt+", encoding="utf-8") as f:453                f.write(attr)454        if sys.platform == "win32":455            # we use ctypes and the windows API to set the hidden attribute on the file456            # only hidden on windows457            import ctypes458            ctypes.windll.kernel32.SetFileAttributesW(459                unicode(self.path_hidden_on_windows), 2460            )461    def tearDown(self):462        import shutil463        shutil.rmtree(self.basepath)464    @ddt.data(465        (None, False),466        ("path_always_visible", False),467        ("path_always_hidden", True),468        ("path_hidden_on_windows", sys.platform == "win32"),469    )470    @ddt.unpack471    def test_is_hidden_path(self, path_id, expected):472        path = getattr(self, path_id) if path_id is not None else None473        self.assertEqual(octoprint.util.is_hidden_path(path), expected)474try:475    from glob import escape  # noqa: F401476except ImportError:477    # no glob.escape - tests for our ported implementation478    @ddt.ddt479    class GlobEscapeTest(unittest.TestCase):480        """481        Ported from Python 3.4482        See https://github.com/python/cpython/commit/fd32fffa5ada8b8be8a65bd51b001d989f99a3d3483        """484        @ddt.data(485            ("abc", "abc"),486            ("[", "[[]"),487            ("?", "[?]"),488            ("*", "[*]"),489            ("[[_/*?*/_]]", "[[][[]_/[*][?][*]/_]]"),490            ("/[[_/*?*/_]]/", "/[[][[]_/[*][?][*]/_]]/"),491        )492        @ddt.unpack493        def test_glob_escape(self, text, expected):494            actual = octoprint.util.glob_escape(text)495            self.assertEqual(actual, expected)496        @ddt.data(497            ("?:?", "?:[?]"),498            ("*:*", "*:[*]"),499            (r"\\?\c:\?", r"\\?\c:\[?]"),500            (r"\\*\*\*", r"\\*\*\[*]"),501            ("//?/c:/?", "//?/c:/[?]"),502            ("//*/*/*", "//*/*/[*]"),503        )504        @ddt.unpack505        @unittest.skipUnless(sys.platform == "win32", "Win32 specific test")506        def test_glob_escape_windows(self, text, expected):507            actual = octoprint.util.glob_escape(text)...

test_temp_utils.py

Source:test_temp_utils.py

...61        self.assertEqual([{"dir": "/run/cloud-init/tmp"}], calls)62    def test_mkstemp_default_non_root(self):63        """mkstemp creates secure tempfile under /tmp for the unprivileged."""64        calls = []65        def fake_mkstemp(*args, **kwargs):66            calls.append(kwargs)67            return "/fake/return/path"68        retval = wrap_and_call(69            "cloudinit.temp_utils",70            {71                "os.getuid": 1000,72                "tempfile.mkstemp": {"side_effect": fake_mkstemp},73                "_TMPDIR": {"new": None},74                "os.path.isdir": True,75            },76            mkstemp,77        )78        self.assertEqual("/fake/return/path", retval)79        self.assertEqual([{"dir": "/tmp"}], calls)80    def test_mkstemp_default_root(self):81        """mkstemp creates a secure tempfile in /run/cloud-init for root."""82        calls = []83        def fake_mkstemp(*args, **kwargs):84            calls.append(kwargs)85            return "/fake/return/path"86        retval = wrap_and_call(87            "cloudinit.temp_utils",88            {89                "os.getuid": 0,90                "tempfile.mkstemp": {"side_effect": fake_mkstemp},91                "_TMPDIR": {"new": None},92                "os.path.isdir": True,93            },94            mkstemp,95        )96        self.assertEqual("/fake/return/path", retval)97        self.assertEqual([{"dir": "/run/cloud-init/tmp"}], calls)...

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.