Add support for Ansible zipped source files

[luenk]

Ansible zipped source support

Scope

It would be great to support Ansible zipped source files to make Ansible module debugging available in PySnooper

Problem

Ansible is using zipped source files on the remote node.
To get access to the source file itself you need to unzip the file during runtime.
The source file will be deleted after execution.

Source path:... /tmp/ansible_my_module_payload_xyz1234/ansible_my_module_payload.zip/ansible/modules/my_module.py
Modified var:.. result = {'changed': True, 'original_message': 'hello', 'message': 'goodbye'}
09:25:25.942249 line       127 SOURCE IS UNAVAILABLE
09:25:25.942279 line       132 SOURCE IS UNAVAILABLE
09:25:25.942303 line       133 SOURCE IS UNAVAILABLE

Solution

I added a condition to tracer.py like the condition for ipython to unzip the source file and get read access.

Source path:... /tmp/ansible_my_module_payload_xyz1234/ansible_my_module_payload.zip/ansible/modules/my_module.py
Modified var:.. result = {'changed': True, 'original_message': 'hello', 'message': 'goodbye'}
09:26:58.498181 line       127     if module.params['name'] == 'fail me':
09:26:58.498211 line       132     if module.params['debug']:
09:26:58.498235 line       133         result['debug'] = pydebug.getvalue()

[cool-RR]

This change looks good to me. Can you add a test for the functionality?

[cool-RR]

Looks good, left one inline comment. Also, there should be a test for the actual usage (extracting source from the zip file).

[luenk]

I added the test cases and changed the string/None handling. Please take a look to the changes.

[cool-RR]

Good tests, quite extensive.

You've got some long lines, please cut lines so no line is longer than 100 characters.

[cool-RR]

Good job. Note that you forgot some breakpoint lines in the code, please delete them. Then make sure your tests pass on Python 2.7 and at least one Python 3.x version. Then I'll merge them.

[cool-RR]

I tried to run the tests but got a failure on test_valid_zipfile:

============================================= FAILURES ============================================= ________________________________________ test_valid_zipfile ________________________________________

    def test_valid_zipfile():
        with mini_toolbox.create_temp_folder(prefix='pysnooper') as folder, \
                                        mini_toolbox.TempSysPathAdder(str(folder)):
            module_name = 'my_valid_zip_module'
            zip_name = 'valid.zip'
            zip_base_path = 'ansible/modules'
            python_file_path = folder / zip_name / zip_base_path / ('%s.py' % (module_name))
            os.makedirs(folder / zip_name / zip_base_path)
            try:
                sys.path.insert(0, str(folder / zip_name))
                content = textwrap.dedent(u'''
                    import pysnooper
                    @pysnooper.snoop(color=False)
                    def f(x):
                        return x
                ''')
                python_file_path.write_text(content)

                module = importlib.import_module('%s.%s' % ('.'.join(zip_base_path.split('/')), \
                                                            module_name))

                with zipfile.ZipFile(folder / 'foo_bar.zip', 'w') as myZipFile:
                    myZipFile.write(folder / zip_name / zip_base_path / ('%s.py' % (module_name)), \
                                    '%s/%s.py' % (zip_base_path, module_name,), \
                                    zipfile.ZIP_DEFLATED)

                python_file_path.unlink()
                folder.joinpath(zip_name).rename(folder.joinpath('%s.delete' % (zip_name)))
                folder.joinpath('foo_bar.zip').rename(folder.joinpath(zip_name))

                with mini_toolbox.OutputCapturer(stdout=False,
                                                 stderr=True) as output_capturer:
                    result = getattr(module, 'f')(7)
                assert result == 7
                output = output_capturer.output

>               assert_output(
                    output,
                    (
                        SourcePathEntry(),
                        VariableEntry(stage='starting'),
                        CallEntry('def f(x):'),
                        LineEntry('return x'),
                        ReturnEntry('return x'),
                        ReturnValueEntry('7'),
                        ElapsedTimeEntry(),
                    )
                )

tests\test_pysnooper.py:1957:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

output = '    Source path:... C:\\Users\\ADMINI~1\\AppData\\Local\\Temp\\pysnooper_mckz2cv\\valid.zip\\ansible\\modules\\my_val...\n    11:24:24.829000 return       5 SOURCE IS UNAVAILABLE\n    Return value:.. 7\n    Elapsed time: 00:00:00.000000\n'
expected_entries = (SourcePathEntry(prefix=''), VariableEntry(stage='starting', prefix=''), CallEntry(source='def f(x):', prefix=''), Lin...(source='return x', prefix=''), ReturnEntry(source='return x', prefix=''), ReturnValueEntry(value='7', prefix=''), ...)
prefix = None, normalize = False

    def assert_output(output, expected_entries, prefix=None, normalize=False):
        lines = tuple(filter(None, output.split('\n')))
        if expected_entries and not lines:
            raise OutputFailure("Output is empty")

        if prefix is not None:
            for line in lines:
                if not line.startswith(prefix):
                    raise OutputFailure(line)

        if normalize:
            verify_normalize(lines, prefix)

        # Filter only entries compatible with the current Python
        filtered_expected_entries = []
        for expected_entry in expected_entries:
            if isinstance(expected_entry, _BaseEntry):
                if expected_entry.is_compatible_with_current_python_version():
                    filtered_expected_entries.append(expected_entry)
            else:
                filtered_expected_entries.append(expected_entry)

        expected_entries_count = len(filtered_expected_entries)
        any_mismatch = False
        result = ''
        template = u'\n{line!s:%s}   {expected_entry}  {arrow}' % max(map(len, lines))
        for expected_entry, line in zip_longest(filtered_expected_entries, lines, fillvalue=""):
            mismatch = not (expected_entry and expected_entry.check(line))
            any_mismatch |= mismatch
            arrow = '<===' * mismatch
            result += template.format(**locals())

        if len(lines) != expected_entries_count:
            result += '\nOutput has {} lines, while we expect {} lines.'.format(
                    len(lines), len(expected_entries))

        if any_mismatch:
>           raise OutputFailure(result)
E           tests.utils.OutputFailure:
E               Source path:... C:\Users\ADMINI~1\AppData\Local\Temp\pysnooper_mckz2cv\valid.zip\ansible\modules\my_valid_zip_module.py   SourcePathEntry(prefix='')
E               Starting var:.. x = 7
                                     VariableEntry(stage='starting', prefix='')
E               11:24:24.829000 call         3 SOURCE IS UNAVAILABLE
                                     CallEntry(source='def f(x):', prefix='')  <===
E               11:24:24.829000 line         5 SOURCE IS UNAVAILABLE
                                     LineEntry(source='return x', prefix='')  <===
E               11:24:24.829000 return       5 SOURCE IS UNAVAILABLE
                                     ReturnEntry(source='return x', prefix='')  <===
E               Return value:.. 7
                                     ReturnValueEntry(value='7', prefix='')
E               Elapsed time: 00:00:00.000000
                                     ElapsedTimeEntry(tolerance=0.2, prefix='')

tests\utils.py:399: OutputFailure
----------- generated html file: file://C:\Users\Administrator\Dropbox\Desktop\foo.html ------------ ===================================== short test summary info ====================================== FAILED tests/test_pysnooper.py::test_valid_zipfile - tests.utils.OutputFailure:
=================================== 1 failed, 77 passed in 2.05s ===================================

[luenk]

Hi,
that is interesting - perhaps there is also an error on Windows.
I did the testing on new installed Linux environments - I will check this.
Thanks for the feedback.
I am still working on the tests. At the moment I am working on the compatibility to python 2.7.
It is not working with the imports in this way and there is also one additional error if a module was already imported.
Test case 1 and case 3 using the same package name.
I will come back to you with working tests. I need a little more time to fix these errors.
Thanks for testing,
Lukas

[luenk]

Hi,
the tests are now python2 and Windows compatible.
Also Windows compatible regex and zip extraction was changed.

The test were running on:
platform darwin -- Python 2.7.18, pytest-4.6.11, py-1.11.0, pluggy-0.13.1
platform darwin -- Python 3.9.12, pytest-7.1.1, pluggy-1.0.0

platform linux2 -- Python 2.7.17, pytest-4.6.11, py-1.11.0, pluggy-0.13.1
platform linux -- Python 3.6.9, pytest-7.0.1, pluggy-1.0.0

platform win32 -- Python 2.7.18, pytest-4.6.11, py-1.11.0, pluggy-0.13.1
platform win32 -- Python 3.10.4, pytest-7.1.1, pluggy-1.0.0

Thanks for all your feedback,
Lukas

[cool-RR]

Good job, merged. The os.makedirs(p) could be replaced by p.mkdir(parents=True), but that shouldn't hold up the PR. I'll make a release.

[luenk]

Thank you so much for merging and all your feedback.
I will take your feedback for future.

Thanks and have a nice weekend,
Lukas