(self, repetitions=10)
| 767 | self.assertTrue(os.path.exists(bar_path)) |
| 768 | |
| 769 | def test_create_tmp(self, repetitions=10): |
| 770 | # Create files in tmp directory |
| 771 | hostname = socket.gethostname() |
| 772 | if '/' in hostname: |
| 773 | hostname = hostname.replace('/', r'\057') |
| 774 | if ':' in hostname: |
| 775 | hostname = hostname.replace(':', r'\072') |
| 776 | pid = os.getpid() |
| 777 | pattern = re.compile(r"(?P<time>\d+)\.M(?P<M>\d{1,6})P(?P<P>\d+)" |
| 778 | r"Q(?P<Q>\d+)\.(?P<host>[^:/]*)") |
| 779 | previous_groups = None |
| 780 | for x in range(repetitions): |
| 781 | tmp_file = self._box._create_tmp() |
| 782 | head, tail = os.path.split(tmp_file.name) |
| 783 | self.assertEqual(head, os.path.abspath(os.path.join(self._path, |
| 784 | "tmp")), |
| 785 | "File in wrong location: '%s'" % head) |
| 786 | match = pattern.match(tail) |
| 787 | self.assertIsNotNone(match, "Invalid file name: '%s'" % tail) |
| 788 | groups = match.groups() |
| 789 | if previous_groups is not None: |
| 790 | self.assertGreaterEqual(int(groups[0]), int(previous_groups[0]), |
| 791 | "Non-monotonic seconds: '%s' before '%s'" % |
| 792 | (previous_groups[0], groups[0])) |
| 793 | if int(groups[0]) == int(previous_groups[0]): |
| 794 | self.assertGreaterEqual(int(groups[1]), int(previous_groups[1]), |
| 795 | "Non-monotonic milliseconds: '%s' before '%s'" % |
| 796 | (previous_groups[1], groups[1])) |
| 797 | self.assertEqual(int(groups[2]), pid, |
| 798 | "Process ID mismatch: '%s' should be '%s'" % |
| 799 | (groups[2], pid)) |
| 800 | self.assertEqual(int(groups[3]), int(previous_groups[3]) + 1, |
| 801 | "Non-sequential counter: '%s' before '%s'" % |
| 802 | (previous_groups[3], groups[3])) |
| 803 | self.assertEqual(groups[4], hostname, |
| 804 | "Host name mismatch: '%s' should be '%s'" % |
| 805 | (groups[4], hostname)) |
| 806 | previous_groups = groups |
| 807 | tmp_file.write(_bytes_sample_message) |
| 808 | tmp_file.seek(0) |
| 809 | self.assertEqual(tmp_file.read(), _bytes_sample_message) |
| 810 | tmp_file.close() |
| 811 | file_count = len(os.listdir(os.path.join(self._path, "tmp"))) |
| 812 | self.assertEqual(file_count, repetitions, |
| 813 | "Wrong file count: '%s' should be '%s'" % |
| 814 | (file_count, repetitions)) |
| 815 | |
| 816 | def test_refresh(self): |
| 817 | # Update the table of contents |
nothing calls this directly
no test coverage detected