← Back to branch summary

~james-w/python-oops/update-readme-dependencies

« back to all changes in this revision

Viewing changes to oops/tests/test_uniquefileallocator.py

  • Committer: Robert Collins
  • Date: 2011-08-15 06:42:14 UTC
  • Revision ID: robertc@robertcollins.net-20110815064214-ncyynjzdwk0ci1a8
Bump to v4, delete disk based storage - moved that to python-oops-datedir-repo - and add core create/publish functionality.

Show diffs side-by-side

added added

removed removed

Lines of Context:
oops/tests/test_uniquefileallocator.py
1
 
# Copyright (c) 2010, 2011, Canonical Ltd
2
 
#
3
 
# This program is free software: you can redistribute it and/or modify
4
 
# it under the terms of the GNU Affero General Public License as published by
5
 
# the Free Software Foundation, either version 3 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU Affero General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU Affero General Public License
14
 
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
15
 
# GNU Affero General Public License version 3 (see the file LICENSE).
16
 
 
17
 
"""Tests for the unique file naming facility."""
18
 
 
19
 
__metaclass__ = type
20
 
 
21
 
import datetime
22
 
import os
23
 
import stat
24
 
 
25
 
from fixtures import TempDir
26
 
import pytz
27
 
import testtools
28
 
 
29
 
from oops.uniquefileallocator import UniqueFileAllocator
30
 
 
31
 
 
32
 
UTC = pytz.timezone('UTC')
33
 
 
34
 
 
35
 
class TestUniqueFileAllocator(testtools.TestCase):
36
 
 
37
 
    def setUp(self):
38
 
        super(TestUniqueFileAllocator, self).setUp()
39
 
        tempdir = self.useFixture(TempDir())
40
 
        self._tempdir = tempdir.path
41
 
 
42
 
    def test_setToken(self):
43
 
        namer = UniqueFileAllocator("/any-old/path/", 'OOPS', 'T')
44
 
        self.assertEqual('T', namer.get_log_infix())
45
 
 
46
 
        # Some scripts will append a string token to the prefix.
47
 
        namer.setToken('CW')
48
 
        self.assertEqual('TCW', namer.get_log_infix())
49
 
 
50
 
        # Some scripts run multiple processes and append a string number
51
 
        # to the prefix.
52
 
        namer.setToken('1')
53
 
        self.assertEqual('T1', namer.get_log_infix())
54
 
 
55
 
    def assertUniqueFileAllocator(self, namer, now, expected_id,
56
 
        expected_last_id, expected_suffix, expected_lastdir):
57
 
        logid, filename = namer.newId(now)
58
 
        self.assertEqual(logid, expected_id)
59
 
        self.assertEqual(filename,
60
 
            os.path.join(namer._output_root, expected_suffix))
61
 
        self.assertEqual(namer._last_serial, expected_last_id)
62
 
        self.assertEqual(namer._last_output_dir,
63
 
            os.path.join(namer._output_root, expected_lastdir))
64
 
 
65
 
    def test_newId(self):
66
 
        # TODO: This should return an id, fileobj instead of a file name, to
67
 
        # reduce races with threads that are slow to use what they asked for,
68
 
        # when combined with configuration changes causing disk scans. That
69
 
        # would also permit using a completely stubbed out file system,
70
 
        # reducing IO in tests that use UniqueFileAllocator (such as all the
71
 
        # pagetests in Launchpad. At that point an interface to obtain a
72
 
        # factory of UniqueFileAllocator's would be useful to parameterise the
73
 
        # entire test suite.
74
 
        namer = UniqueFileAllocator(self._tempdir, 'OOPS', 'T')
75
 
        # first name of the day
76
 
        self.assertUniqueFileAllocator(namer,
77
 
            datetime.datetime(2006, 04, 01, 00, 30, 00, tzinfo=UTC),
78
 
            'OOPS-91T1', 1, '2006-04-01/01800.T1', '2006-04-01')
79
 
        # second name of the day
80
 
        self.assertUniqueFileAllocator(namer,
81
 
            datetime.datetime(2006, 04, 01, 12, 00, 00, tzinfo=UTC),
82
 
            'OOPS-91T2', 2, '2006-04-01/43200.T2', '2006-04-01')
83
 
 
84
 
        # first name of the following day sets a new dir and the id starts
85
 
        # over.
86
 
        self.assertUniqueFileAllocator(namer,
87
 
            datetime.datetime(2006, 04, 02, 00, 30, 00, tzinfo=UTC),
88
 
            'OOPS-92T1', 1, '2006-04-02/01800.T1', '2006-04-02')
89
 
 
90
 
        # Setting a token inserts the token into the filename.
91
 
        namer.setToken('YYY')
92
 
        logid, filename = namer.newId(
93
 
            datetime.datetime(2006, 04, 02, 00, 30, 00, tzinfo=UTC))
94
 
        self.assertEqual(logid, 'OOPS-92TYYY2')
95
 
 
96
 
        # Setting a type controls the log id:
97
 
        namer.setToken('')
98
 
        namer._log_type = "PROFILE"
99
 
        logid, filename = namer.newId(
100
 
            datetime.datetime(2006, 04, 02, 00, 30, 00, tzinfo=UTC))
101
 
        self.assertEqual(logid, 'PROFILE-92T3')
102
 
 
103
 
        # Native timestamps are not permitted - UTC only.
104
 
        now = datetime.datetime(2006, 04, 02, 00, 30, 00)
105
 
        self.assertRaises(ValueError, namer.newId, now)
106
 
 
107
 
    def test_changeErrorDir(self):
108
 
        """Test changing the log output dir."""
109
 
        namer = UniqueFileAllocator(self._tempdir, 'OOPS', 'T')
110
 
 
111
 
        # First an id in the original error directory.
112
 
        self.assertUniqueFileAllocator(namer,
113
 
            datetime.datetime(2006, 04, 01, 00, 30, 00, tzinfo=UTC),
114
 
            'OOPS-91T1', 1, '2006-04-01/01800.T1', '2006-04-01')
115
 
 
116
 
        # UniqueFileAllocator uses the _output_root attribute to get the
117
 
        # current output directory.
118
 
        new_output_dir = self.useFixture(TempDir()).path
119
 
        namer._output_root = new_output_dir
120
 
 
121
 
        # Now an id on the same day, in the new directory.
122
 
        now = datetime.datetime(2006, 04, 01, 12, 00, 00, tzinfo=UTC)
123
 
        log_id, filename = namer.newId(now)
124
 
 
125
 
        # Since it's a new directory, with no previous logs, the id is 1
126
 
        # again, rather than 2.
127
 
        self.assertEqual(log_id, 'OOPS-91T1')
128
 
        self.assertEqual(namer._last_serial, 1)
129
 
        self.assertEqual(namer._last_output_dir,
130
 
            os.path.join(new_output_dir, '2006-04-01'))
131
 
 
132
 
    def test_findHighestSerial(self):
133
 
        namer = UniqueFileAllocator(self._tempdir, "OOPS", "T")
134
 
        # Creates the dir using now as the timestamp.
135
 
        output_dir = namer.output_dir()
136
 
        # write some files, in non-serial order.
137
 
        open(os.path.join(output_dir, '12343.T1'), 'w').close()
138
 
        open(os.path.join(output_dir, '12342.T2'), 'w').close()
139
 
        open(os.path.join(output_dir, '12345.T3'), 'w').close()
140
 
        open(os.path.join(output_dir, '1234567.T0010'), 'w').close()
141
 
        open(os.path.join(output_dir, '12346.A42'), 'w').close()
142
 
        open(os.path.join(output_dir, '12346.B100'), 'w').close()
143
 
        # The namer should figure out the right highest serial.
144
 
        self.assertEqual(namer._findHighestSerial(output_dir), 10)
145
 
 
146
 
    def test_output_dir_permission(self):
147
 
        # Set up default dir creation mode to rwx------.
148
 
        umask_permission = stat.S_IRWXG | stat.S_IRWXO
149
 
        old_umask = os.umask(umask_permission)
150
 
        namer = UniqueFileAllocator(self._tempdir, "OOPS", "T")
151
 
        output_dir = namer.output_dir()
152
 
        st = os.stat(output_dir)
153
 
        # Permission we want here is: rwxr-xr-x
154
 
        wanted_permission = (
155
 
            stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH |
156
 
            stat.S_IXOTH)
157
 
        # Get only the permission bits for this directory.
158
 
        dir_permission = stat.S_IMODE(st.st_mode)
159
 
        self.assertEqual(dir_permission, wanted_permission)
160
 
        # Restore the umask to the original value.
161
 
        ignored = os.umask(old_umask)