FilesExpand file tree

 master

/

test.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

125 lines (103 loc) · 4.15 KB

 master

/

test.py

Copy path

Top

File metadata and controls

• Code
• Blame

125 lines (103 loc) · 4.15 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

122

123

124

125

# 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.

import os

import re

import shlex

import subprocess

import testtools

import six

from tempest.lib.cli import output_parser

from tempest.lib import exceptions

COMMON_DIR = os.path.dirname(os.path.abspath(__file__))

FUNCTIONAL_DIR = os.path.normpath(os.path.join(COMMON_DIR, '..'))

ROOT_DIR = os.path.normpath(os.path.join(FUNCTIONAL_DIR, '..'))

EXAMPLE_DIR = os.path.join(ROOT_DIR, 'examples')

def execute(cmd, fail_ok=False, merge_stderr=False):

"""Executes specified command for the given action."""

cmdlist = shlex.split(cmd)

result = ''

result_err = ''

stdout = subprocess.PIPE

stderr = subprocess.STDOUT if merge_stderr else subprocess.PIPE

proc = subprocess.Popen(cmdlist, stdout=stdout, stderr=stderr)

result, result_err = proc.communicate()

result = result.decode('utf-8')

if not fail_ok and proc.returncode != 0:

raise exceptions.CommandFailed(proc.returncode, cmd, result,

result_err)

return result

class TestCase(testtools.TestCase):

delimiter_line = re.compile('^\+\-[\+\-]+\-\+$')

@classmethod

def openstack(cls, cmd, fail_ok=False):

"""Executes openstackclient command for the given action."""

return execute('openstack ' + cmd, fail_ok=fail_ok)

@classmethod

def get_openstack_configuration_value(cls, configuration):

opts = cls.get_show_opts([configuration])

return cls.openstack('configuration show ' + opts)

@classmethod

def get_show_opts(cls, fields=[]):

return ' -f value ' + ' '.join(['-c ' + it for it in fields])

@classmethod

def get_list_opts(cls, headers=[]):

opts = ' -f csv '

opts = opts + ' '.join(['-c ' + it for it in headers])

return opts

@classmethod

def assertOutput(cls, expected, actual):

if expected != actual:

raise Exception(expected + ' != ' + actual)

@classmethod

def assertInOutput(cls, expected, actual):

if expected not in actual:

raise Exception(expected + ' not in ' + actual)

def assert_table_structure(self, items, field_names):

"""Verify that all items have keys listed in field_names."""

for item in items:

for field in field_names:

self.assertIn(field, item)

def assert_show_fields(self, items, field_names):

"""Verify that all items have keys listed in field_names."""

for item in items:

for key in six.iterkeys(item):

self.assertIn(key, field_names)

def assert_show_structure(self, items, field_names):

"""Verify that all field_names listed in keys of all items."""

if isinstance(items, list):

o = {}

for d in items:

o.update(d)

else:

o = items

item_keys = o.keys()

for field in field_names:

self.assertIn(field, item_keys)

def parse_show_as_object(self, raw_output):

"""Return a dict with values parsed from cli output."""

items = self.parse_show(raw_output)

o = {}

for item in items:

o.update(item)

return o

def parse_show(self, raw_output):

"""Return list of dicts with item values parsed from cli output."""

items = []

table_ = output_parser.table(raw_output)

for row in table_['values']:

item = {}

item[row[0]] = row[1]

items.append(item)

return items

def parse_listing(self, raw_output):

"""Return list of dicts with basic item parsed from cli output."""

return output_parser.listing(raw_output)