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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
|
"""An implementation of a mock GCS data source for nsscache."""
import datetime
import io
import unittest
from unittest import mock
from nss_cache.maps import group
from nss_cache.maps import passwd
from nss_cache.maps import shadow
from nss_cache.util import file_formats
from nss_cache.util import timestamps
try:
from nss_cache.sources import gcssource
except Exception as e:
raise unittest.SkipTest("`gcssource` unabled to be imported: {}".format(e))
class TestGcsSource(unittest.TestCase):
def setUp(self):
super(TestGcsSource, self).setUp()
self.config = {
"passwd_object": "PASSWD_OBJ",
"group_object": "GROUP_OBJ",
"bucket": "TEST_BUCKET",
}
def testDefaultConfiguration(self):
source = gcssource.GcsFilesSource({})
self.assertEqual(source.conf["bucket"], gcssource.GcsFilesSource.BUCKET)
self.assertEqual(
source.conf["passwd_object"], gcssource.GcsFilesSource.PASSWD_OBJECT
)
def testOverrideDefaultConfiguration(self):
source = gcssource.GcsFilesSource(self.config)
self.assertEqual(source.conf["bucket"], "TEST_BUCKET")
self.assertEqual(source.conf["passwd_object"], "PASSWD_OBJ")
self.assertEqual(source.conf["group_object"], "GROUP_OBJ")
class TestPasswdUpdateGetter(unittest.TestCase):
def setUp(self):
super(TestPasswdUpdateGetter, self).setUp()
self.updater = gcssource.PasswdUpdateGetter()
def testGetParser(self):
self.assertIsInstance(
self.updater.GetParser(), file_formats.FilesPasswdMapParser
)
def testCreateMap(self):
self.assertIsInstance(self.updater.CreateMap(), passwd.PasswdMap)
class TestShadowUpdateGetter(unittest.TestCase):
def setUp(self):
super(TestShadowUpdateGetter, self).setUp()
self.updater = gcssource.ShadowUpdateGetter()
def testGetParser(self):
self.assertIsInstance(
self.updater.GetParser(), file_formats.FilesShadowMapParser
)
def testCreateMap(self):
self.assertIsInstance(self.updater.CreateMap(), shadow.ShadowMap)
def testShadowGetUpdatesWithContent(self):
mock_client = mock.Mock()
mock_bucket = mock_client.bucket.return_value
mock_blob = mock_bucket.get_blob.return_value
mock_blob.open.return_value = io.StringIO(
"""usera:x:::::::
userb:x:::::::
"""
)
mock_blob.updated = datetime.datetime.now()
result = self.updater.GetUpdates(mock_client, "test-bucket", "passwd", None)
self.assertEqual(len(result), 2)
mock_bucket.get_blob.assert_called_with("passwd")
mock_client.bucket.assert_called_with("test-bucket")
def testShadowGetUpdatesSinceAfterUpdatedTime(self):
mock_client = mock.Mock()
mock_bucket = mock_client.bucket.return_value
mock_blob = mock_bucket.get_blob.return_value
now = datetime.datetime.now()
mock_blob.updated = now
result = self.updater.GetUpdates(
mock_client,
"test-bucket",
"passwd",
timestamps.FromDateTimeToTimestamp(now + datetime.timedelta(days=1)),
)
self.assertEqual(len(result), 0)
mock_bucket.get_blob.assert_called_with("passwd")
mock_client.bucket.assert_called_with("test-bucket")
class TestGroupUpdateGetter(unittest.TestCase):
def setUp(self):
super(TestGroupUpdateGetter, self).setUp()
self.updater = gcssource.GroupUpdateGetter()
def testGetParser(self):
self.assertIsInstance(
self.updater.GetParser(), file_formats.FilesGroupMapParser
)
def testCreateMap(self):
self.assertIsInstance(self.updater.CreateMap(), group.GroupMap)
if __name__ == "__main__":
unittest.main()
|
