Revision a9d68e40 test/ganeti.utils.io_unittest.py
| b/test/ganeti.utils.io_unittest.py | ||
|---|---|---|
| 234 | 234 |
|
| 235 | 235 |
class TestWriteFile(unittest.TestCase): |
| 236 | 236 |
def setUp(self): |
| 237 |
self.tmpdir = None |
|
| 237 | 238 |
self.tfile = tempfile.NamedTemporaryFile() |
| 238 | 239 |
self.did_pre = False |
| 239 | 240 |
self.did_post = False |
| 240 | 241 |
self.did_write = False |
| 241 | 242 |
|
| 243 |
def tearDown(self): |
|
| 244 |
if self.tmpdir: |
|
| 245 |
shutil.rmtree(self.tmpdir) |
|
| 246 |
|
|
| 242 | 247 |
def markPre(self, fd): |
| 243 | 248 |
self.did_pre = True |
| 244 | 249 |
|
| ... | ... | |
| 260 | 265 |
self.assertRaises(errors.ProgrammerError, utils.WriteFile, |
| 261 | 266 |
self.tfile.name, data="test", atime=0) |
| 262 | 267 |
|
| 263 |
def testCalls(self): |
|
| 264 |
utils.WriteFile(self.tfile.name, fn=self.markWrite, |
|
| 265 |
prewrite=self.markPre, postwrite=self.markPost) |
|
| 268 |
def testPreWrite(self): |
|
| 269 |
utils.WriteFile(self.tfile.name, data="", prewrite=self.markPre) |
|
| 266 | 270 |
self.assertTrue(self.did_pre) |
| 271 |
self.assertFalse(self.did_post) |
|
| 272 |
self.assertFalse(self.did_write) |
|
| 273 |
|
|
| 274 |
def testPostWrite(self): |
|
| 275 |
utils.WriteFile(self.tfile.name, data="", postwrite=self.markPost) |
|
| 276 |
self.assertFalse(self.did_pre) |
|
| 267 | 277 |
self.assertTrue(self.did_post) |
| 278 |
self.assertFalse(self.did_write) |
|
| 279 |
|
|
| 280 |
def testWriteFunction(self): |
|
| 281 |
utils.WriteFile(self.tfile.name, fn=self.markWrite) |
|
| 282 |
self.assertFalse(self.did_pre) |
|
| 283 |
self.assertFalse(self.did_post) |
|
| 268 | 284 |
self.assertTrue(self.did_write) |
| 269 | 285 |
|
| 270 | 286 |
def testDryRun(self): |
| ... | ... | |
| 293 | 309 |
finally: |
| 294 | 310 |
os.close(fd) |
| 295 | 311 |
|
| 312 |
def testNoLeftovers(self): |
|
| 313 |
self.tmpdir = tempfile.mkdtemp() |
|
| 314 |
self.assertEqual(utils.WriteFile(utils.PathJoin(self.tmpdir, "test"), |
|
| 315 |
data="abc"), |
|
| 316 |
None) |
|
| 317 |
self.assertEqual(os.listdir(self.tmpdir), ["test"]) |
|
| 318 |
|
|
| 319 |
def testFailRename(self): |
|
| 320 |
self.tmpdir = tempfile.mkdtemp() |
|
| 321 |
target = utils.PathJoin(self.tmpdir, "target") |
|
| 322 |
os.mkdir(target) |
|
| 323 |
self.assertRaises(OSError, utils.WriteFile, target, data="abc") |
|
| 324 |
self.assertTrue(os.path.isdir(target)) |
|
| 325 |
self.assertEqual(os.listdir(self.tmpdir), ["target"]) |
|
| 326 |
self.assertFalse(os.listdir(target)) |
|
| 327 |
|
|
| 328 |
def testFailRenameDryRun(self): |
|
| 329 |
self.tmpdir = tempfile.mkdtemp() |
|
| 330 |
target = utils.PathJoin(self.tmpdir, "target") |
|
| 331 |
os.mkdir(target) |
|
| 332 |
self.assertEqual(utils.WriteFile(target, data="abc", dry_run=True), None) |
|
| 333 |
self.assertTrue(os.path.isdir(target)) |
|
| 334 |
self.assertEqual(os.listdir(self.tmpdir), ["target"]) |
|
| 335 |
self.assertFalse(os.listdir(target)) |
|
| 336 |
|
|
| 337 |
def testBackup(self): |
|
| 338 |
self.tmpdir = tempfile.mkdtemp() |
|
| 339 |
testfile = utils.PathJoin(self.tmpdir, "test") |
|
| 340 |
|
|
| 341 |
self.assertEqual(utils.WriteFile(testfile, data="foo", backup=True), None) |
|
| 342 |
self.assertEqual(utils.ReadFile(testfile), "foo") |
|
| 343 |
self.assertEqual(os.listdir(self.tmpdir), ["test"]) |
|
| 344 |
|
|
| 345 |
# Write again |
|
| 346 |
assert os.path.isfile(testfile) |
|
| 347 |
self.assertEqual(utils.WriteFile(testfile, data="bar", backup=True), None) |
|
| 348 |
self.assertEqual(utils.ReadFile(testfile), "bar") |
|
| 349 |
self.assertEqual(len(glob.glob("%s.backup*" % testfile)), 1)
|
|
| 350 |
self.assertTrue("test" in os.listdir(self.tmpdir))
|
|
| 351 |
self.assertEqual(len(os.listdir(self.tmpdir)), 2) |
|
| 352 |
|
|
| 353 |
# Write again as dry-run |
|
| 354 |
assert os.path.isfile(testfile) |
|
| 355 |
self.assertEqual(utils.WriteFile(testfile, data="000", backup=True, |
|
| 356 |
dry_run=True), |
|
| 357 |
None) |
|
| 358 |
self.assertEqual(utils.ReadFile(testfile), "bar") |
|
| 359 |
self.assertEqual(len(glob.glob("%s.backup*" % testfile)), 1)
|
|
| 360 |
self.assertTrue("test" in os.listdir(self.tmpdir))
|
|
| 361 |
self.assertEqual(len(os.listdir(self.tmpdir)), 2) |
|
| 362 |
|
|
| 296 | 363 |
|
| 297 | 364 |
class TestFileID(testutils.GanetiTestCase): |
| 298 | 365 |
def testEquality(self): |
Also available in: Unified diff