#26 - Fix a bug that causes output to be repeated several times when …

…invoking pytest with -s flag (#28)

#26 - Fix a bug that causes output to be repeated several times when invoking pytest with '-s' flag. The requirement on memory_profiler has been upgraded in order to enforce only one iteration for collecting memory usage

.circleci/config.yml

@@ -27,7 +27,7 @@ shared: &shared
                 conda init bash
                 conda activate ci
                 mkdir test-results
-                pytest --junit-xml=test-results/junit.xml
+                pytest --junit-xml=test-results/junit.xml -s
         - store_test_results:
             path: test-results
         - store_artifacts:

pytest_monitor/pytest_monitor.py

@@ -118,12 +118,12 @@ def pytest_runtest_call(item):
     setattr(item, 'monitor_component', getattr(item.module, 'pytest_monitor_component', ''))
 
 
+@pytest.hookimpl
 def pytest_pyfunc_call(pyfuncitem):
     """
     Core sniffer logic. We encapsulate the test function in a sniffer function to collect
     memory results.
     """
-
     def wrapped_function():
         try:
             funcargs = pyfuncitem.funcargs
@@ -135,10 +135,12 @@ def wrapped_function():
             return e
 
     def prof():
-        m = memory_profiler.memory_usage((wrapped_function, ()), max_usage=True, retval=True)
+        m = memory_profiler.memory_usage((wrapped_function, ()),
+                                         max_iterations=1, max_usage=True, retval=True)
         if isinstance(m[1], BaseException):  # Do we have any outcome?
             raise m[1]
-        setattr(pyfuncitem, 'mem_usage', m[0])
+        memuse = m[0][0] if type(m[0]) is list else m[0]
+        setattr(pyfuncitem, 'mem_usage', memuse)
         setattr(pyfuncitem, 'monitor_results', True)
     prof()
     return True

pytest_monitor/session.py

@@ -63,7 +63,7 @@ def compute_info(self, description, tags):
         h.update(description.encode())
         self.__session = h.hexdigest()
         # From description + tags to JSON format
-        d=dict()
+        d = dict()
         if description:
             d['description'] = description
         for tag in tags:
@@ -72,7 +72,7 @@ def compute_info(self, description, tags):
                 d[_tag_info[0]] = _tag_info[1]
             else:
                 for sub_tag in tag:
-                    _tag_info = subtag.split('=', 1)
+                    _tag_info = sub_tag.split('=', 1)
                     d[_tag_info[0]] = _tag_info[1]
         description = json.dumps(d)
         # Now get memory usage base and create the database
@@ -111,7 +111,8 @@ def prepare(self):
         def dummy():
             return True
 
-        self.__mem_usage_base = memory_profiler.memory_usage((dummy,), max_usage=True)
+        memuse = memory_profiler.memory_usage((dummy,), max_iterations=1, max_usage=True)
+        self.__mem_usage_base = memuse[0] if type(memuse) is list else memuse
 
     def add_test_info(self, item, item_path, item_variant, item_loc, kind, component,
                       item_start_time, total_time, user_time, kernel_time, mem_usage):

requirements.txt

@@ -1,3 +1,4 @@
 psutil
-memory_profiler
+memory_profiler >= 0.58
 pytest
+requests

tests/test_monitor.py

@@ -253,3 +253,25 @@ def test_that():
 
     # make sure that that we get a '0' exit code for the testsuite
     result.assert_outcomes(passed=2)
+
+
+def test_monitor_basic_output(testdir):
+    """Make sure that pytest-monitor does not repeat captured output (issue #26)."""
+
+    # create a temporary pytest test module
+    testdir.makepyfile("""
+        def test_it():
+            print('Hello World')
+    """)
+
+    wrn = 'pytest-monitor: No storage specified but monitoring is requested. Disabling monitoring.'
+    with pytest.warns(UserWarning, match=wrn):
+        # run pytest with the following cmd args
+        result = testdir.runpytest('--no-db', '-s', '-vv')
+
+    # fnmatch_lines does an assertion internally
+    result.stdout.fnmatch_lines(['*::test_it Hello World*'])
+    assert "Hello World" != result.stdout.get_lines_after('*Hello World')[0]
+
+    # make sure that that we get a '0' exit code for the testsuite
+    result.assert_outcomes(passed=1)