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
|
###
# ============LICENSE_START=======================================================
# Simulator
# ================================================================================
# Copyright (C) 2019 Nokia. All rights reserved.
# ================================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============LICENSE_END=========================================================
###
import unittest
from unittest import mock
from werkzeug.datastructures import FileStorage
from yang_loader_server import YangLoaderHelper, YangModelServer
class TestYangLoaderHelper(unittest.TestCase):
def test_should_save_file_and_return_path(self):
helper = YangLoaderHelper()
mocked_file = mock.Mock(FileStorage)
mocked_file.filename = "sample"
path = helper.save_file(mocked_file)
self.assertEqual(path, "/tmp/sample")
mocked_file.save.assert_called_once_with("/tmp/sample")
@mock.patch('yang_loader_server.check_output')
def test_should_install_new_yang_model(self, mocked_output):
helper = YangLoaderHelper()
helper.install_new_model("path")
mocked_output.assert_called_with(
['sysrepoctl', '--install', '--yang=path',
'--owner=netconf:nogroup', '--permissions=777'],
stderr=-2, universal_newlines=True)
@mock.patch('yang_loader_server.check_output')
def test_should_delete_yang_model(self, mocked_output):
helper = YangLoaderHelper()
helper.uninstall_a_model("modelName")
mocked_output.assert_called_with(
['sysrepoctl', '--uninstall', '--module=modelName'],
stderr=-2, universal_newlines=True)
@mock.patch('yang_loader_server.check_output')
def test_should_set_default_configuration(self, mocked_output):
helper = YangLoaderHelper()
helper.set_default_configuration("samplePath", "sampleModuleName")
mocked_output.assert_called_with(
['sysrepocfg', '--import=samplePath', '--datastore=startup',
'--format=xml', '--level=3', 'sampleModuleName'],
stderr=-2, universal_newlines=True)
@mock.patch('yang_loader_server.subprocess.Popen')
@mock.patch('yang_loader_server.check_output')
def test_should_verify_change_listener_for_model_properly(self, mocked_output, mocked_popen):
helper = YangLoaderHelper()
helper.start_change_listener_for_model("sampleModule")
mocked_output.assert_called_with(
['pgrep', '-f', '/opt/dev/netopeer-change-saver/bin/netopeer-change-saver sampleModule kafka1 config'],
stderr=-2, universal_newlines=True)
@mock.patch('yang_loader_server.check_output')
def test_should_raise_exception_when_error_occurred_in_output(self,
mocked_output):
helper = YangLoaderHelper()
mocked_output.return_value = "abcd ERR"
with self.assertRaises(RuntimeError) as context:
helper._run_bash_command("sample command")
self.assertEqual('abcd ERR', str(context.exception))
class TestYangModelServer(unittest.TestCase):
def __init__(self, methodName='runTest'):
super().__init__(methodName)
self._mocked_file = mock.Mock(FileStorage)
def test_should_properly_apply_and_start_new_model(self):
with mock.patch.object(YangModelServer, '_parse_request',
new=self._mock_request):
helper = mock.Mock(YangLoaderHelper)
helper.save_file.return_value = "sampleFile"
server = YangModelServer(helper)
server.post()
self.assertEqual(helper.save_file.call_count, 2)
helper.install_new_model.assert_called_once_with('sampleFile')
helper.set_default_configuration.assert_called_once_with(
'sampleFile', 'sampleModuleName')
helper.start_change_listener_for_model.assert_called_once_with('sampleModuleName')
def _mock_request(self):
return {
'yangModel': self._mocked_file,
'initialConfig': self._mocked_file,
'moduleName': "sampleModuleName"
}
|
