← Back to branch summary

~niedbalski/charms/trusty/rabbitmq-server/fix-lp-1489053

« back to all changes in this revision

Viewing changes to tests/charmhelpers/core/fstab.py

  • Committer: Liam Young
  • Date: 2015-09-09 13:12:47 UTC
  • mfrom: (110.1.4 rabbitmq-server)
  • Revision ID: liam.young@canonical.com-20150909131247-16hxw74o91c57kpg
[1chb1n, r=gnuoy] Refactor amulet tests, deprecate old tests

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/charmhelpers/core/fstab.py
1
 
#!/usr/bin/env python
2
 
# -*- coding: utf-8 -*-
3
 
 
4
 
# Copyright 2014-2015 Canonical Limited.
5
 
#
6
 
# This file is part of charm-helpers.
7
 
#
8
 
# charm-helpers is free software: you can redistribute it and/or modify
9
 
# it under the terms of the GNU Lesser General Public License version 3 as
10
 
# published by the Free Software Foundation.
11
 
#
12
 
# charm-helpers is distributed in the hope that it will be useful,
13
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
14
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15
 
# GNU Lesser General Public License for more details.
16
 
#
17
 
# You should have received a copy of the GNU Lesser General Public License
18
 
# along with charm-helpers.  If not, see <http://www.gnu.org/licenses/>.
19
 
 
20
 
import io
21
 
import os
22
 
 
23
 
__author__ = 'Jorge Niedbalski R. <jorge.niedbalski@canonical.com>'
24
 
 
25
 
 
26
 
class Fstab(io.FileIO):
27
 
    """This class extends file in order to implement a file reader/writer
28
 
    for file `/etc/fstab`
29
 
    """
30
 
 
31
 
    class Entry(object):
32
 
        """Entry class represents a non-comment line on the `/etc/fstab` file
33
 
        """
34
 
        def __init__(self, device, mountpoint, filesystem,
35
 
                     options, d=0, p=0):
36
 
            self.device = device
37
 
            self.mountpoint = mountpoint
38
 
            self.filesystem = filesystem
39
 
 
40
 
            if not options:
41
 
                options = "defaults"
42
 
 
43
 
            self.options = options
44
 
            self.d = int(d)
45
 
            self.p = int(p)
46
 
 
47
 
        def __eq__(self, o):
48
 
            return str(self) == str(o)
49
 
 
50
 
        def __str__(self):
51
 
            return "{} {} {} {} {} {}".format(self.device,
52
 
                                              self.mountpoint,
53
 
                                              self.filesystem,
54
 
                                              self.options,
55
 
                                              self.d,
56
 
                                              self.p)
57
 
 
58
 
    DEFAULT_PATH = os.path.join(os.path.sep, 'etc', 'fstab')
59
 
 
60
 
    def __init__(self, path=None):
61
 
        if path:
62
 
            self._path = path
63
 
        else:
64
 
            self._path = self.DEFAULT_PATH
65
 
        super(Fstab, self).__init__(self._path, 'rb+')
66
 
 
67
 
    def _hydrate_entry(self, line):
68
 
        # NOTE: use split with no arguments to split on any
69
 
        #       whitespace including tabs
70
 
        return Fstab.Entry(*filter(
71
 
            lambda x: x not in ('', None),
72
 
            line.strip("\n").split()))
73
 
 
74
 
    @property
75
 
    def entries(self):
76
 
        self.seek(0)
77
 
        for line in self.readlines():
78
 
            line = line.decode('us-ascii')
79
 
            try:
80
 
                if line.strip() and not line.strip().startswith("#"):
81
 
                    yield self._hydrate_entry(line)
82
 
            except ValueError:
83
 
                pass
84
 
 
85
 
    def get_entry_by_attr(self, attr, value):
86
 
        for entry in self.entries:
87
 
            e_attr = getattr(entry, attr)
88
 
            if e_attr == value:
89
 
                return entry
90
 
        return None
91
 
 
92
 
    def add_entry(self, entry):
93
 
        if self.get_entry_by_attr('device', entry.device):
94
 
            return False
95
 
 
96
 
        self.write((str(entry) + '\n').encode('us-ascii'))
97
 
        self.truncate()
98
 
        return entry
99
 
 
100
 
    def remove_entry(self, entry):
101
 
        self.seek(0)
102
 
 
103
 
        lines = [l.decode('us-ascii') for l in self.readlines()]
104
 
 
105
 
        found = False
106
 
        for index, line in enumerate(lines):
107
 
            if line.strip() and not line.strip().startswith("#"):
108
 
                if self._hydrate_entry(line) == entry:
109
 
                    found = True
110
 
                    break
111
 
 
112
 
        if not found:
113
 
            return False
114
 
 
115
 
        lines.remove(line)
116
 
 
117
 
        self.seek(0)
118
 
        self.write(''.join(lines).encode('us-ascii'))
119
 
        self.truncate()
120
 
        return True
121
 
 
122
 
    @classmethod
123
 
    def remove_by_mountpoint(cls, mountpoint, path=None):
124
 
        fstab = cls(path=path)
125
 
        entry = fstab.get_entry_by_attr('mountpoint', mountpoint)
126
 
        if entry:
127
 
            return fstab.remove_entry(entry)
128
 
        return False
129
 
 
130
 
    @classmethod
131
 
    def add(cls, device, mountpoint, filesystem, options=None, path=None):
132
 
        return cls(path=path).add_entry(Fstab.Entry(device,
133
 
                                                    mountpoint, filesystem,
134
 
                                                    options=options))