Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

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

"""Tests for certbot.plugins.storage.PluginStorage""" 

import json 

import mock 

import os 

import unittest 

 

from certbot import errors 

 

from certbot.plugins import common 

from certbot.tests import util as test_util 

 

class PluginStorageTest(test_util.ConfigTestCase): 

"""Test for certbot.plugins.storage.PluginStorage""" 

 

def setUp(self): 

super(PluginStorageTest, self).setUp() 

self.plugin_cls = common.Installer 

os.mkdir(self.config.config_dir) 

with mock.patch("certbot.reverter.util"): 

self.plugin = self.plugin_cls(config=self.config, name="mockplugin") 

 

def test_load_errors_cant_read(self): 

with open(os.path.join(self.config.config_dir, 

".pluginstorage.json"), "w") as fh: 

fh.write("dummy") 

# When unable to read file that exists 

mock_open = mock.mock_open() 

mock_open.side_effect = IOError 

self.plugin.storage.storagepath = os.path.join(self.config.config_dir, 

".pluginstorage.json") 

with mock.patch("six.moves.builtins.open", mock_open): 

with mock.patch('os.path.isfile', return_value=True): 

with mock.patch("certbot.reverter.util"): 

self.assertRaises(errors.PluginStorageError, 

self.plugin.storage._load) # pylint: disable=protected-access 

 

def test_load_errors_empty(self): 

with open(os.path.join(self.config.config_dir, ".pluginstorage.json"), "w") as fh: 

fh.write('') 

with mock.patch("certbot.plugins.storage.logger.debug") as mock_log: 

# Should not error out but write a debug log line instead 

with mock.patch("certbot.reverter.util"): 

nocontent = self.plugin_cls(self.config, "mockplugin") 

self.assertRaises(KeyError, 

nocontent.storage.fetch, "value") 

self.assertTrue(mock_log.called) 

self.assertTrue("no values loaded" in mock_log.call_args[0][0]) 

 

def test_load_errors_corrupted(self): 

with open(os.path.join(self.config.config_dir, 

".pluginstorage.json"), "w") as fh: 

fh.write('invalid json') 

with mock.patch("certbot.plugins.storage.logger.error") as mock_log: 

with mock.patch("certbot.reverter.util"): 

corrupted = self.plugin_cls(self.config, "mockplugin") 

self.assertRaises(errors.PluginError, 

corrupted.storage.fetch, 

"value") 

self.assertTrue("is corrupted" in mock_log.call_args[0][0]) 

 

def test_save_errors_cant_serialize(self): 

with mock.patch("certbot.plugins.storage.logger.error") as mock_log: 

# Set data as something that can't be serialized 

self.plugin.storage._initialized = True # pylint: disable=protected-access 

self.plugin.storage.storagepath = "/tmp/whatever" 

self.plugin.storage._data = self.plugin_cls # pylint: disable=protected-access 

self.assertRaises(errors.PluginStorageError, 

self.plugin.storage.save) 

self.assertTrue("Could not serialize" in mock_log.call_args[0][0]) 

 

def test_save_errors_unable_to_write_file(self): 

mock_open = mock.mock_open() 

mock_open.side_effect = IOError 

with mock.patch("os.open", mock_open): 

with mock.patch("certbot.plugins.storage.logger.error") as mock_log: 

self.plugin.storage._data = {"valid": "data"} # pylint: disable=protected-access 

self.plugin.storage._initialized = True # pylint: disable=protected-access 

self.assertRaises(errors.PluginStorageError, 

self.plugin.storage.save) 

self.assertTrue("Could not write" in mock_log.call_args[0][0]) 

 

def test_save_uninitialized(self): 

with mock.patch("certbot.reverter.util"): 

self.assertRaises(errors.PluginStorageError, 

self.plugin_cls(self.config, "x").storage.save) 

 

def test_namespace_isolation(self): 

with mock.patch("certbot.reverter.util"): 

plugin1 = self.plugin_cls(self.config, "first") 

plugin2 = self.plugin_cls(self.config, "second") 

plugin1.storage.put("first_key", "first_value") 

self.assertRaises(KeyError, 

plugin2.storage.fetch, "first_key") 

self.assertRaises(KeyError, 

plugin2.storage.fetch, "first") 

self.assertEqual(plugin1.storage.fetch("first_key"), "first_value") 

 

 

def test_saved_state(self): 

self.plugin.storage.put("testkey", "testvalue") 

# Write to disk 

self.plugin.storage.save() 

with mock.patch("certbot.reverter.util"): 

another = self.plugin_cls(self.config, "mockplugin") 

self.assertEqual(another.storage.fetch("testkey"), "testvalue") 

 

with open(os.path.join(self.config.config_dir, 

".pluginstorage.json"), 'r') as fh: 

psdata = fh.read() 

psjson = json.loads(psdata) 

self.assertTrue("mockplugin" in psjson.keys()) 

self.assertEqual(len(psjson), 1) 

self.assertEqual(psjson["mockplugin"]["testkey"], "testvalue") 

 

 

if __name__ == "__main__": 

unittest.main() # pragma: no cover