-
Notifications
You must be signed in to change notification settings - Fork 85
/
fssize.py
155 lines (113 loc) · 4.66 KB
/
fssize.py
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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
# fssize.py
# Filesystem size gathering classes.
#
# Copyright (C) 2015 Red Hat, Inc.
#
# This copyrighted material is made available to anyone wishing to use,
# modify, copy, or redistribute it subject to the terms and conditions of
# the GNU General Public License v.2, or (at your option) any later version.
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY expressed or implied, including the implied warranties of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General
# Public License for more details. You should have received a copy of the
# GNU General Public License along with this program; if not, write to the
# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA. Any Red Hat trademarks that are incorporated in the
# source code or documentation are not subject to the GNU General Public
# License and may only be used or replicated with the express permission of
# Red Hat, Inc.
#
# Red Hat Author(s): Anne Mulhern <amulhern@redhat.com>
import abc
from collections import namedtuple
from six import add_metaclass
from ..errors import FSError
from ..size import Size
from .. import util
from . import availability
from . import fstask
from . import task
_tags = ("count", "size")
_Tags = namedtuple("_Tags", _tags)
@add_metaclass(abc.ABCMeta)
class FSSize(fstask.FSTask):
""" An abstract class that represents size information extraction. """
description = "current filesystem size"
tags = abc.abstractproperty(
doc="Strings used for extracting components of size.")
# TASK methods
@property
def _availability_errors(self):
return []
@property
def depends_on(self):
return [self.fs._info]
# IMPLEMENTATION methods
def do_task(self):
""" Returns the size of the filesystem.
:returns: the size of the filesystem
:rtype: :class:`~.size.Size`
:raises FSError: on failure
"""
error_msgs = self.availability_errors
if error_msgs:
raise FSError("\n".join(error_msgs))
if self.fs._current_info is None:
raise FSError("No info available for size computation.")
# Setup initial values
values = {}
for k in _tags:
values[k] = None
# Attempt to set values from info
for line in (l.strip() for l in self.fs._current_info.splitlines()):
key = next((k for k in _tags if line.startswith(getattr(self.tags, k))), None)
if not key:
continue
if values[key] is not None:
raise FSError("found two matches for key %s" % key)
# Look for last numeric value in matching line
fields = line.split()
fields.reverse()
for field in fields:
try:
values[key] = int(field)
break
except ValueError:
continue
# Raise an error if a value is missing
missing = next((k for k in _tags if values[k] is None), None)
if missing is not None:
raise FSError("Failed to parse info for %s." % missing)
return values["count"] * Size(values["size"])
class Ext2FSSize(FSSize):
tags = _Tags(size="Block size:", count="Block count:")
class JFSSize(FSSize):
tags = _Tags(size="Physical block size:", count="Aggregate size:")
class NTFSSize(FSSize):
tags = _Tags(size="Cluster Size:", count="Volume Size in Clusters:")
class ReiserFSSize(FSSize):
tags = _Tags(size="Blocksize:", count="Count of blocks on the device:")
class XFSSize(FSSize):
tags = _Tags(size="blocksize =", count="dblocks =")
class TmpFSSize(task.BasicApplication, fstask.FSTask):
description = "current filesystem size"
ext = availability.DF_APP
@property
def _size_command(self):
return [str(self.ext), self.fs.system_mountpoint, "--output=size"]
def do_task(self):
error_msgs = self.availability_errors
if error_msgs:
raise FSError("\n".join(error_msgs))
try:
(ret, out) = util.run_program_and_capture_output(self._size_command)
if ret:
raise FSError("Failed to execute command %s." % self._size_command)
except OSError:
raise FSError("Failed to execute command %s." % self._size_command)
lines = out.splitlines()
if len(lines) != 2 or lines[0].strip() != "1K-blocks":
raise FSError("Failed to parse output of command %s." % self._size_command)
return Size("%s KiB" % lines[1])
class UnimplementedFSSize(fstask.UnimplementedFSTask):
pass