summaryrefslogtreecommitdiffstats
path: root/meta/lib/oeqa/runtime/cases/rpm.py
blob: 8e18b426f8b38cfd0ee0dbe20de8f46ca8a06eff (plain)
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
#
# SPDX-License-Identifier: MIT
#

import os
import fnmatch
import time

from oeqa.runtime.case import OERuntimeTestCase
from oeqa.core.decorator.depends import OETestDepends
from oeqa.core.decorator.data import skipIfDataVar
from oeqa.runtime.decorator.package import OEHasPackage
from oeqa.core.utils.path import findFile

class RpmBasicTest(OERuntimeTestCase):

    @OEHasPackage(['rpm'])
    @OETestDepends(['ssh.SSHTest.test_ssh'])
    def test_rpm_help(self):
        status, output = self.target.run('rpm --help')
        msg = 'status and output: %s and %s' % (status, output)
        self.assertEqual(status, 0, msg=msg)

    @OETestDepends(['rpm.RpmBasicTest.test_rpm_help'])
    def test_rpm_query(self):
        status, output = self.target.run('ls /var/lib/rpm/')
        if status != 0:
            self.skipTest('No /var/lib/rpm on target')
        status, output = self.target.run('rpm -q rpm')
        msg = 'status and output: %s and %s' % (status, output)
        self.assertEqual(status, 0, msg=msg)

    @OETestDepends(['rpm.RpmBasicTest.test_rpm_query'])
    def test_rpm_query_nonroot(self):

        def set_up_test_user(u):
            status, output = self.target.run('id -u %s' % u)
            if status:
                status, output = self.target.run('useradd %s' % u)
                msg = 'Failed to create new user: %s' % output
                self.assertTrue(status == 0, msg=msg)

        def exec_as_test_user(u):
            status, output = self.target.run('su -c id %s' % u)
            msg = 'Failed to execute as new user'
            self.assertTrue("({0})".format(u) in output, msg=msg)

            status, output = self.target.run('su -c "rpm -qa" %s ' % u)
            msg = 'status: %s. Cannot run rpm -qa: %s' % (status, output)
            self.assertEqual(status, 0, msg=msg)

        def check_no_process_for_user(u):
            _, output = self.target.run(self.tc.target_cmds['ps'])
            if u + ' ' in output:
                return False
            else:
                return True

        def unset_up_test_user(u):
            # ensure no test1 process in running
            timeout = time.time() + 30
            while time.time() < timeout:
                if check_no_process_for_user(u):
                    break
                else:
                    time.sleep(1)
            status, output = self.target.run('userdel -r %s' % u)
            msg = 'Failed to erase user: %s' % output
            self.assertTrue(status == 0, msg=msg)

        tuser = 'test1'

        try:
            set_up_test_user(tuser)
            exec_as_test_user(tuser)
        finally:
            unset_up_test_user(tuser)


class RpmInstallRemoveTest(OERuntimeTestCase):

    @classmethod
    def setUpClass(cls):
        pkgarch = cls.td['TUNE_PKGARCH'].replace('-', '_')
        rpmdir = os.path.join(cls.tc.td['DEPLOY_DIR'], 'rpm', pkgarch)
        # Pick base-passwd-doc as a test file to get installed, because it's small
        # and it will always be built for standard targets
        rpm_doc = 'base-passwd-doc-*.%s.rpm' % pkgarch
        if not os.path.exists(rpmdir):
            return
        for f in fnmatch.filter(os.listdir(rpmdir), rpm_doc):
            cls.test_file = os.path.join(rpmdir, f)
        cls.dst = '/tmp/base-passwd-doc.rpm'

    @OETestDepends(['rpm.RpmBasicTest.test_rpm_query'])
    def test_rpm_install(self):
        self.tc.target.copyTo(self.test_file, self.dst)
        status, output = self.target.run('rpm -ivh /tmp/base-passwd-doc.rpm')
        msg = 'Failed to install base-passwd-doc package: %s' % output
        self.assertEqual(status, 0, msg=msg)
        self.tc.target.run('rm -f %s' % self.dst)

    @OETestDepends(['rpm.RpmInstallRemoveTest.test_rpm_install'])
    def test_rpm_remove(self):
        status,output = self.target.run('rpm -e base-passwd-doc')
        msg = 'Failed to remove base-passwd-doc package: %s' % output
        self.assertEqual(status, 0, msg=msg)

    @OETestDepends(['rpm.RpmInstallRemoveTest.test_rpm_remove'])
    def test_check_rpm_install_removal_log_file_size(self):
        """
        Summary:     Check that rpm writes into /var/log/messages
        Expected:    There should be some RPM prefixed entries in the above file.
        Product:     BSPs
        Author:      Alexandru Georgescu <alexandru.c.georgescu@intel.com>
        Author:      Alexander Kanavin <alex.kanavin@gmail.com>
        AutomatedBy: Daniel Istrate <daniel.alexandrux.istrate@intel.com>
        """
        db_files_cmd = 'ls /var/lib/rpm/__db.*'
        check_log_cmd = "grep RPM /var/log/messages | wc -l"

        # Make sure that some database files are under /var/lib/rpm as '__db.xxx'
        status, output = self.target.run(db_files_cmd)
        msg =  'Failed to find database files under /var/lib/rpm/ as __db.xxx'
        self.assertEqual(0, status, msg=msg)

        self.tc.target.copyTo(self.test_file, self.dst)

        # Remove the package just in case
        self.target.run('rpm -e base-passwd-doc')

        # Install/Remove a package 10 times
        for i in range(10):
            status, output = self.target.run('rpm -ivh /tmp/base-passwd-doc.rpm')
            msg = 'Failed to install base-passwd-doc package. Reason: {}'.format(output)
            self.assertEqual(0, status, msg=msg)

            status, output = self.target.run('rpm -e base-passwd-doc')
            msg = 'Failed to remove base-passwd-doc package. Reason: {}'.format(output)
            self.assertEqual(0, status, msg=msg)

        self.tc.target.run('rm -f %s' % self.dst)

        # if using systemd this should ensure all entries are flushed to /var
        status, output = self.target.run("journalctl --sync")
        # Get the amount of entries in the log file
        status, output = self.target.run(check_log_cmd)
        msg = 'Failed to get the final size of the log file.'
        self.assertEqual(0, status, msg=msg)

        # Check that there's enough of them
        self.assertGreaterEqual(int(output), 80,
                                   'Cound not find sufficient amount of rpm entries in /var/log/messages, found {} entries'.format(output))