Merge branch 'master' into setup-teardown-deferreds

Author: jelmer

NEWS

@@ -11,6 +11,17 @@ Changes
 
 Fully migrate away from Launchpad to GitHub.
 
+Improvements
+------------
+
+* Support binary contents in ``FileContains`` matcher.
+  (Jelmer Vernooij, #538)
+
+* Allow stream=None to be passed to various TestResult
+  classes that now support verbosity; fixes a regression
+from 2.8.0.
+  (Jelmer Vernooij)
+
 2.8.1
 ~~~~~
 

tests/matchers/test_filesystem.py

@@ -175,6 +175,41 @@ def test_does_not_contain(self):
             Equals(mismatch.describe()),
         )
 
+    def test_binary_content(self):
+        tempdir = self.mkdtemp()
+        filename = os.path.join(tempdir, "binary_file")
+        binary_data = b"\x00\x01\x02\x03\xff\xfe"
+        with open(filename, "wb") as f:
+            f.write(binary_data)
+        self.assertThat(filename, FileContains(binary_data))
+
+    def test_binary_content_mismatch(self):
+        tempdir = self.mkdtemp()
+        filename = os.path.join(tempdir, "binary_file")
+        with open(filename, "wb") as f:
+            f.write(b"\x00\x01\x02")
+        mismatch = FileContains(b"\xff\xfe\xfd").match(filename)
+        self.assertThat(
+            Equals(b"\xff\xfe\xfd").match(b"\x00\x01\x02").describe(),
+            Equals(mismatch.describe()),
+        )
+
+    def test_text_with_encoding(self):
+        tempdir = self.mkdtemp()
+        filename = os.path.join(tempdir, "utf8_file")
+        text_data = "Hello 世界!"
+        with open(filename, "w", encoding="utf-8") as f:
+            f.write(text_data)
+        self.assertThat(filename, FileContains(text_data, encoding="utf-8"))
+
+    def test_text_default_encoding(self):
+        tempdir = self.mkdtemp()
+        filename = os.path.join(tempdir, "text_file")
+        text_data = "Hello World!"
+        with open(filename, "w") as f:
+            f.write(text_data)
+        self.assertThat(filename, FileContains(text_data))
+
 
 class TestTarballContains(TestCase, PathHelpers):
     def test_match(self):

tests/test_testresult.py

@@ -2078,6 +2078,65 @@ def run_tests():
             ),
         )
 
+    def test_none_stream_is_accepted(self):
+        """TextTestResult should accept None as stream for backward compatibility."""
+        result = TextTestResult(None)
+        test = make_test()
+        # Should not raise AttributeError
+        result.startTestRun()
+        result.startTest(test)
+        result.addSuccess(test)
+        result.stopTest(test)
+        result.stopTestRun()
+
+    def test_none_stream_with_failure(self):
+        """TextTestResult with None stream should handle failures."""
+        result = TextTestResult(None)
+        test = make_failing_test()
+        # Should not raise AttributeError
+        test.run(result)
+        self.assertEqual(1, len(result.failures))
+
+    def test_verbosity_zero_produces_no_output(self):
+        """TextTestResult with verbosity=0 should produce no per-test output."""
+        stream = io.StringIO()
+        result = TextTestResult(stream, verbosity=0)
+        test = make_test()
+        result.startTestRun()
+        result.startTest(test)
+        result.addSuccess(test)
+        result.stopTest(test)
+        # Get output up to stopTestRun (which still outputs summary)
+        output_before_stop = stream.getvalue()
+        # Should only have "Tests running...\n" from startTestRun
+        self.assertEqual("Tests running...\n", output_before_stop)
+
+    def test_verbosity_zero_allows_subclass_control(self):
+        """Subclasses can use verbosity=0 to control their own output."""
+
+        class CustomResult(TextTestResult):
+            def __init__(self, stream):
+                super().__init__(stream, verbosity=0)
+
+            def addSuccess(self, test, details=None):
+                super().addSuccess(test, details)
+                self.stream.write("CUSTOM_SUCCESS_MARKER\n")
+
+        stream = io.StringIO()
+        result = CustomResult(stream)
+        test = make_test()
+        result.startTestRun()
+        result.startTest(test)
+        result.addSuccess(test)
+        result.stopTest(test)
+        output = stream.getvalue()
+        # Should have custom output but not parent's dot or "ok"
+        self.assertIn("CUSTOM_SUCCESS_MARKER\n", output)
+        self.assertNotIn("ok\n", output)
+        # Count dots - "Tests running..." has 3 dots, should not have a 4th
+        # from the success indicator
+        self.assertEqual(output.count("."), 3)
+
 
 class TestThreadSafeForwardingResult(TestCase):
     """Tests for `TestThreadSafeForwardingResult`."""

testtools/matchers/_filesystem.py

@@ -94,7 +94,7 @@ def match(self, path):
 class FileContains(Matcher):
     """Matches if the given file has the specified contents."""
 
-    def __init__(self, contents=None, matcher=None):
+    def __init__(self, contents=None, matcher=None, encoding=None):
         """Construct a ``FileContains`` matcher.
 
         Can be used in a basic mode where the file contents are compared for
@@ -103,9 +103,14 @@ def __init__(self, contents=None, matcher=None):
         matched against an arbitrary matcher (by passing ``matcher`` instead).
 
         :param contents: If specified, match the contents of the file with
-            these contents.
+            these contents. If bytes, the file will be opened in binary mode.
+            If str, the file will be opened in text mode using the specified
+            encoding (or the default encoding if not specified).
         :param matcher: If specified, match the contents of the file against
             this matcher.
+        :param encoding: Optional text encoding to use when opening the file
+            in text mode. Only used when contents is a str (or when using a
+            matcher for text content). Defaults to the system default encoding.
         """
         if contents == matcher is None:
             raise AssertionError("Must provide one of `contents` or `matcher`.")
@@ -115,19 +120,23 @@ def __init__(self, contents=None, matcher=None):
             )
         if matcher is None:
             self.matcher = Equals(contents)
+            self._binary_mode = isinstance(contents, bytes)
         else:
             self.matcher = matcher
+            self._binary_mode = False
+        self.encoding = encoding
 
     def match(self, path):
         mismatch = PathExists().match(path)
         if mismatch is not None:
             return mismatch
-        f = open(path)
-        try:
-            actual_contents = f.read()
-            return self.matcher.match(actual_contents)
-        finally:
-            f.close()
+        if self._binary_mode:
+            with open(path, "rb") as f:
+                actual_contents: bytes | str = f.read()
+        else:
+            with open(path, encoding=self.encoding) as f:
+                actual_contents = f.read()
+        return self.matcher.match(actual_contents)
 
     def __str__(self):
         return f"File at path exists and contains {self.matcher}"

testtools/testresult/real.py

@@ -1244,6 +1244,8 @@ def _delta_to_float(self, a_timedelta, precision):
         )
 
     def _show_list(self, label, error_list):
+        if self.stream is None:
+            return
         for test, output in error_list:
             self.stream.write(self.sep1)
             self.stream.write(f"{label}: {test.id()}\n")
@@ -1252,69 +1254,76 @@ def _show_list(self, label, error_list):
 
     def startTest(self, test):
         super().startTest(test)
-        if self.verbosity >= 2:
+        if self.stream is not None and self.verbosity >= 2:
             self.stream.write(f"{test.id()} ... ")
             self.stream.flush()
 
     def addSuccess(self, test, details=None):
         super().addSuccess(test, details=details)
-        if self.verbosity == 1:
-            self.stream.write(".")
-            self.stream.flush()
-            self._progress_printed = True
-        elif self.verbosity >= 2:
-            self.stream.write("ok\n")
-            self.stream.flush()
+        if self.stream is not None:
+            if self.verbosity == 1:
+                self.stream.write(".")
+                self.stream.flush()
+                self._progress_printed = True
+            elif self.verbosity >= 2:
+                self.stream.write("ok\n")
+                self.stream.flush()
 
     def addError(self, test, err=None, details=None):
         super().addError(test, err=err, details=details)
-        if self.verbosity == 1:
-            self.stream.write("E")
-            self.stream.flush()
-        elif self.verbosity >= 2:
-            self.stream.write("ERROR\n")
-            self.stream.flush()
+        if self.stream is not None:
+            if self.verbosity == 1:
+                self.stream.write("E")
+                self.stream.flush()
+            elif self.verbosity >= 2:
+                self.stream.write("ERROR\n")
+                self.stream.flush()
 
     def addFailure(self, test, err=None, details=None):
         super().addFailure(test, err=err, details=details)
-        if self.verbosity == 1:
-            self.stream.write("F")
-            self.stream.flush()
-        elif self.verbosity >= 2:
-            self.stream.write("FAIL\n")
-            self.stream.flush()
+        if self.stream is not None:
+            if self.verbosity == 1:
+                self.stream.write("F")
+                self.stream.flush()
+            elif self.verbosity >= 2:
+                self.stream.write("FAIL\n")
+                self.stream.flush()
 
     def addSkip(self, test, reason=None, details=None):
         super().addSkip(test, reason=reason, details=details)
-        if self.verbosity == 1:
-            self.stream.write("s")
-            self.stream.flush()
-        elif self.verbosity >= 2:
-            self.stream.write(f"skipped {reason!r}\n")
-            self.stream.flush()
+        if self.stream is not None:
+            if self.verbosity == 1:
+                self.stream.write("s")
+                self.stream.flush()
+            elif self.verbosity >= 2:
+                self.stream.write(f"skipped {reason!r}\n")
+                self.stream.flush()
 
     def addExpectedFailure(self, test, err=None, details=None):
         super().addExpectedFailure(test, err=err, details=details)
-        if self.verbosity == 1:
-            self.stream.write("x")
-            self.stream.flush()
-        elif self.verbosity >= 2:
-            self.stream.write("expected failure\n")
-            self.stream.flush()
+        if self.stream is not None:
+            if self.verbosity == 1:
+                self.stream.write("x")
+                self.stream.flush()
+            elif self.verbosity >= 2:
+                self.stream.write("expected failure\n")
+                self.stream.flush()
 
     def addUnexpectedSuccess(self, test, details=None):
         super().addUnexpectedSuccess(test, details=details)
-        if self.verbosity == 1:
-            self.stream.write("u")
-            self.stream.flush()
-        elif self.verbosity >= 2:
-            self.stream.write("unexpected success\n")
-            self.stream.flush()
+        if self.stream is not None:
+            if self.verbosity == 1:
+                self.stream.write("u")
+                self.stream.flush()
+            elif self.verbosity >= 2:
+                self.stream.write("unexpected success\n")
+                self.stream.flush()
 
     def startTestRun(self):
         super().startTestRun()
         self.__start = self._now()
-        self.stream.write("Tests running...\n")
+        if self.stream is not None:
+            self.stream.write("Tests running...\n")
 
     def stopTestRun(self):
         if self.testsRun != 1:
@@ -1324,31 +1333,33 @@ def stopTestRun(self):
         stop = self._now()
         self._show_list("ERROR", self.errors)
         self._show_list("FAIL", self.failures)
-        for test in self.unexpectedSuccesses:
+        if self.stream is not None:
+            for test in self.unexpectedSuccesses:
+                self.stream.write(
+                    f"{self.sep1}UNEXPECTED SUCCESS: {test.id()}\n{self.sep2}"
+                )
+            # Add newline(s) before summary
+            # If we printed progress indicators (dots), add extra newline
+            if self._progress_printed:
+                self.stream.write("\n\n")
+            else:
+                self.stream.write("\n")
             self.stream.write(
-                f"{self.sep1}UNEXPECTED SUCCESS: {test.id()}\n{self.sep2}"
-            )
-        # Add newline(s) before summary
-        # If we printed progress indicators (dots), add extra newline
-        if self._progress_printed:
-            self.stream.write("\n\n")
-        else:
-            self.stream.write("\n")
-        self.stream.write(
-            f"Ran {self.testsRun} test{plural} in "
-            f"{self._delta_to_float(stop - self.__start, 3):.3f}s\n"
-        )
-        if self.wasSuccessful():
-            self.stream.write("OK\n")
-        else:
-            self.stream.write("FAILED (")
-            details = []
-            failure_count = sum(
-                len(x) for x in (self.failures, self.errors, self.unexpectedSuccesses)
+                f"Ran {self.testsRun} test{plural} in "
+                f"{self._delta_to_float(stop - self.__start, 3):.3f}s\n"
             )
-            details.append(f"failures={failure_count}")
-            self.stream.write(", ".join(details))
-            self.stream.write(")\n")
+            if self.wasSuccessful():
+                self.stream.write("OK\n")
+            else:
+                self.stream.write("FAILED (")
+                details = []
+                failure_count = sum(
+                    len(x)
+                    for x in (self.failures, self.errors, self.unexpectedSuccesses)
+                )
+                details.append(f"failures={failure_count}")
+                self.stream.write(", ".join(details))
+                self.stream.write(")\n")
         super().stopTestRun()