File: env_file_watcher_test.py

package info (click to toggle)
python-mitogen 0.3.0~rc1-4
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 3,240 kB
  • sloc: python: 19,899; sh: 91; perl: 19; ansic: 18; makefile: 13
file content (74 lines) | stat: -rw-r--r-- 1,980 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import os
import sys
import tempfile

import mock
import unittest2
import testlib

from mitogen.core import b
import ansible_mitogen.runner


klass = ansible_mitogen.runner.EnvironmentFileWatcher
environb = getattr(os, 'environb', os.environ)


class WatcherTest(testlib.TestCase):
    def setUp(self):
        self.original_env = environb.copy()
        self.tf = tempfile.NamedTemporaryFile()

    def tearDown(self):
        self.tf.close()
        environb.clear()
        environb.update(self.original_env)

    def test_missing_file(self):
        # just ensure it doesn't crash
        watcher = klass('/nonexistent')
        watcher.check()

    def test_file_becomes_missing(self):
        # just ensure it doesn't crash
        watcher = klass(self.tf.name)
        watcher.check()
        os.unlink(self.tf.name)
        watcher.check()
        open(self.tf.name,'wb').close()

    def test_key_deleted(self):
        environb[b('SOMEKEY')] = b('123')
        self.tf.write(b('SOMEKEY=123\n'))
        self.tf.flush()
        watcher = klass(self.tf.name)
        self.tf.seek(0)
        self.tf.truncate(0)
        watcher.check()
        self.assertTrue(b('SOMEKEY') not in environb)

    def test_key_added(self):
        watcher = klass(self.tf.name)
        self.tf.write(b('SOMEKEY=123\n'))
        self.tf.flush()
        watcher.check()
        self.assertEqual(environb[b('SOMEKEY')], b('123'))

    def test_key_shadowed_nuchange(self):
        environb[b('SOMEKEY')] = b('234')
        self.tf.write(b('SOMEKEY=123\n'))
        self.tf.flush()
        watcher = klass(self.tf.name)
        watcher.check()
        self.assertEqual(environb[b('SOMEKEY')], b('234'))

    def test_binary_key_added(self):
        watcher = klass(self.tf.name)
        self.tf.write(b('SOMEKEY=\xff\xff\xff\n'))
        self.tf.flush()
        watcher.check()
        self.assertEqual(environb[b('SOMEKEY')], b('\xff\xff\xff'))


if __name__ == '__main__':
    unittest2.main()