300-miner-basic-ops.py 12.3 KB
Newer Older
1
#!/usr/bin/env python3
Ivan Frade's avatar
Ivan Frade committed
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

# Copyright (C) 2010, Nokia (ivan.frade@nokia.com)
#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This library is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this library; if not, write to the
# Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
# Boston, MA  02110-1301, USA.

#
# TODO:
#     These tests are for files... we need to write them for folders!
#
"""
Monitor a test directory and copy/move/remove/update files and folders there.
Check the basic data of the files is updated accordingly in tracker.
"""
import os
import shutil
import time

32
import unittest as ut
33
from common.utils.helpers import log
34
from common.utils.minertest import CommonTrackerMinerTest
Ivan Frade's avatar
Ivan Frade committed
35

36
NFO_DOCUMENT = 'http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#Document'
37

38

Ivan Frade's avatar
Ivan Frade committed
39 40 41 42
class MinerCrawlTest (CommonTrackerMinerTest):
    """
    Test cases to check if miner is able to monitor files that are created, deleted or moved
    """
43 44 45

    def __get_text_documents(self):
        return self.tracker.query("""
Ivan Frade's avatar
Ivan Frade committed
46 47
          SELECT ?url WHERE {
              ?u a nfo:TextDocument ;
48
                 nie:url ?url.
Ivan Frade's avatar
Ivan Frade committed
49 50 51
          }
          """)

52 53
    def __get_parent_urn(self, filepath):
        result = self.tracker.query("""
54 55 56 57
          SELECT nfo:belongsToContainer(?u) WHERE {
              ?u a nfo:FileDataObject ;
                 nie:url \"%s\" .
          }
58 59
          """ % (self.uri(filepath)))
        self.assertEqual(len(result), 1)
60 61
        return result[0][0]

62 63
    def __get_file_urn(self, filepath):
        result = self.tracker.query("""
64 65 66 67
          SELECT ?u WHERE {
              ?u a nfo:FileDataObject ;
                 nie:url \"%s\" .
          }
68 69
          """ % (self.uri(filepath)))
        self.assertEqual(len(result), 1)
70 71
        return result[0][0]

Ivan Frade's avatar
Ivan Frade committed
72 73 74
    """
    Boot the miner with the correct configuration and check everything is fine
    """
75 76

    def test_01_initial_crawling(self):
Ivan Frade's avatar
Ivan Frade committed
77 78 79
        """
        The precreated files and folders should be there
        """
80 81 82 83 84 85
        result = self.__get_text_documents()
        self.assertEqual(len(result), 3)
        unpacked_result = [r[0] for r in result]
        self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
86

Ivan Frade's avatar
Ivan Frade committed
87 88 89
        # We don't check (yet) folders, because Applications module is injecting results


90 91 92 93
# class copy(TestUpdate):
# FIXME all tests in one class because the miner-fs restarting takes some time (~5 sec)
# Maybe we can move the miner-fs initialization to setUpModule and then move these
# tests to different classes
94

95 96

    def test_02_copy_from_unmonitored_to_monitored(self):
Ivan Frade's avatar
Ivan Frade committed
97 98 99 100
        """
        Copy an file from unmonitored directory to monitored directory
        and verify if data base is updated accordingly
        """
101 102 103
        source = os.path.join(self.workdir, "test-no-monitored", "file0.txt")
        dest = os.path.join(self.workdir, "test-monitored", "file0.txt")
        shutil.copyfile(source, dest)
104

105
        dest_id, dest_urn = self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(dest))
Ivan Frade's avatar
Ivan Frade committed
106

107
        # verify if miner indexed this file.
108 109 110 111 112 113 114
        result = self.__get_text_documents()
        self.assertEqual(len(result), 4)
        unpacked_result = [r[0] for r in result]
        self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/file0.txt"), unpacked_result)
Ivan Frade's avatar
Ivan Frade committed
115 116

        # Clean the new file so the test directory is as before
117 118 119
        log("Remove and wait")
        os.remove(dest)
        self.system.store.await_resource_deleted(NFO_DOCUMENT, dest_id)
Ivan Frade's avatar
Ivan Frade committed
120

121
    def test_03_copy_from_monitored_to_unmonitored(self):
Ivan Frade's avatar
Ivan Frade committed
122 123
        """
        Copy an file from a monitored location to an unmonitored location
124
        Nothing should change
Ivan Frade's avatar
Ivan Frade committed
125 126 127
        """

        # Copy from monitored to unmonitored
128 129 130
        source = os.path.join(self.workdir, "test-monitored", "file1.txt")
        dest = os.path.join(self.workdir, "test-no-monitored", "file1.txt")
        shutil.copyfile(source, dest)
Ivan Frade's avatar
Ivan Frade committed
131

132
        time.sleep(1)
Ivan Frade's avatar
Ivan Frade committed
133
        # Nothing changed
134 135 136 137 138 139
        result = self.__get_text_documents()
        self.assertEqual(len(result), 3, "Results:" + str(result))
        unpacked_result = [r[0] for r in result]
        self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
140

Ivan Frade's avatar
Ivan Frade committed
141
        # Clean the file
142
        os.remove(dest)
Ivan Frade's avatar
Ivan Frade committed
143

144
    def test_04_copy_from_monitored_to_monitored(self):
Ivan Frade's avatar
Ivan Frade committed
145 146 147
        """
        Copy a file between monitored directories
        """
148 149 150
        source = os.path.join(self.workdir, "test-monitored", "file1.txt")
        dest = os.path.join(self.workdir, "test-monitored", "dir1", "dir2", "file-test04.txt")
        shutil.copyfile(source, dest)
151

152
        dest_id, dest_urn = self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(dest))
Ivan Frade's avatar
Ivan Frade committed
153

154 155 156 157 158 159 160
        result = self.__get_text_documents()
        self.assertEqual(len(result), 4)
        unpacked_result = [r[0] for r in result]
        self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/dir2/file-test04.txt"), unpacked_result)
161

Ivan Frade's avatar
Ivan Frade committed
162
        # Clean the file
163 164 165
        os.remove(dest)
        self.system.store.await_resource_deleted(NFO_DOCUMENT, dest_id)
        self.assertEqual(3, self.tracker.count_instances("nfo:TextDocument"))
Ivan Frade's avatar
Ivan Frade committed
166

167
    @ut.skip("https://gitlab.gnome.org/GNOME/tracker-miners/issues/56")
168
    def test_05_move_from_unmonitored_to_monitored(self):
Ivan Frade's avatar
Ivan Frade committed
169 170 171
        """
        Move a file from unmonitored to monitored directory
        """
172 173 174 175 176 177 178 179 180 181 182 183
        source = os.path.join(self.workdir, "test-no-monitored", "file0.txt")
        dest = os.path.join(self.workdir, "test-monitored", "dir1", "file-test05.txt")
        shutil.move(source, dest)
        dest_id, dest_urn = self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(dest))

        result = self.__get_text_documents()
        self.assertEqual(len(result), 4)
        unpacked_result = [r[0] for r in result]
        self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/file-test05.txt"), unpacked_result)
184

Ivan Frade's avatar
Ivan Frade committed
185
        # Clean the file
186 187 188
        os.remove(dest)
        self.system.store.await_resource_deleted(NFO_DOCUMENT, dest_id)
        self.assertEqual(3, self.tracker.count_instances("nfo:TextDocument"))
Ivan Frade's avatar
Ivan Frade committed
189 190

## """ move operation and tracker-miner response test cases """
191
# class move(TestUpdate):
Ivan Frade's avatar
Ivan Frade committed
192

193
    def test_06_move_from_monitored_to_unmonitored(self):
Ivan Frade's avatar
Ivan Frade committed
194 195 196
        """
        Move a file from monitored to unmonitored directory
        """
197 198
        source = self.path("test-monitored/dir1/file2.txt")
        dest = self.path("test-no-monitored/file2.txt")
199 200 201
        source_id = self.system.store.get_resource_id(self.uri(source))
        shutil.move(source, dest)
        self.system.store.await_resource_deleted(NFO_DOCUMENT, source_id)
Ivan Frade's avatar
Ivan Frade committed
202

203 204 205 206 207
        result = self.__get_text_documents()
        self.assertEqual(len(result), 2)
        unpacked_result = [r[0] for r in result]
        self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
208

Ivan Frade's avatar
Ivan Frade committed
209
        # Restore the file
210 211 212
        shutil.move(dest, source)
        self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(source))
        self.assertEqual(3, self.tracker.count_instances("nfo:TextDocument"))
Ivan Frade's avatar
Ivan Frade committed
213

214
    def test_07_move_from_monitored_to_monitored(self):
Ivan Frade's avatar
Ivan Frade committed
215 216 217
        """
        Move a file between monitored directories
        """
218

219 220
        source = self.path("test-monitored/dir1/file2.txt")
        dest = self.path("test-monitored/file2.txt")
221

222
        resource_id = self.tracker.get_resource_id(url=self.uri(source))
223

224 225 226
        source_dir_urn = self.__get_file_urn(os.path.dirname(source))
        parent_before = self.__get_parent_urn(source)
        self.assertEqual(source_dir_urn, parent_before)
227

228
        shutil.move(source, dest)
229
        self.tracker.await_property_changed(NFO_DOCUMENT, resource_id, 'nie:url')
Ivan Frade's avatar
Ivan Frade committed
230

231 232
        # Checking fix for NB#214413: After a move operation, nfo:belongsToContainer
        # should be changed to the new one
233 234 235 236 237 238 239 240 241 242 243
        dest_dir_urn = self.__get_file_urn(os.path.dirname(dest))
        parent_after = self.__get_parent_urn(dest)
        self.assertNotEqual(parent_before, parent_after)
        self.assertEqual(dest_dir_urn, parent_after)

        result = self.__get_text_documents()
        self.assertEqual(len(result), 3)
        unpacked_result = [r[0] for r in result]
        self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/file2.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
Ivan Frade's avatar
Ivan Frade committed
244 245

        # Restore the file
246
        shutil.move(dest, source)
247
        self.tracker.await_property_changed(NFO_DOCUMENT, resource_id, 'nie:url')
Ivan Frade's avatar
Ivan Frade committed
248

249 250 251 252
        result = self.__get_text_documents()
        self.assertEqual(len(result), 3)
        unpacked_result = [r[0] for r in result]
        self.assertIn(self.uri("test-monitored/dir1/file2.txt"), unpacked_result)
Ivan Frade's avatar
Ivan Frade committed
253

254
    def test_08_deletion_single_file(self):
Ivan Frade's avatar
Ivan Frade committed
255 256 257
        """
        Delete one of the files
        """
258
        victim = self.path("test-monitored/dir1/file2.txt")
259 260 261
        victim_id = self.system.store.get_resource_id(self.uri(victim))
        os.remove(victim)
        self.system.store.await_resource_deleted(NFO_DOCUMENT, victim_id)
Ivan Frade's avatar
Ivan Frade committed
262

263 264 265 266 267
        result = self.__get_text_documents()
        self.assertEqual(len(result), 2)
        unpacked_result = [r[0] for r in result]
        self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(self.uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
Ivan Frade's avatar
Ivan Frade committed
268 269

        # Restore the file
270 271 272 273
        f = open(victim, "w")
        f.write("Don't panic, everything is fine")
        f.close()
        self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(victim))
Ivan Frade's avatar
Ivan Frade committed
274

275
    def test_09_deletion_directory(self):
Ivan Frade's avatar
Ivan Frade committed
276 277 278
        """
        Delete a directory
        """
279
        victim = self.path("test-monitored/dir1")
280 281
        victim_id = self.system.store.get_resource_id(self.uri(victim))
        shutil.rmtree(victim)
282

283 284 285
        file_inside_victim_url = self.uri(os.path.join(victim, "file2.txt"))
        file_inside_victim_id = self.system.store.get_resource_id(file_inside_victim_url)
        self.system.store.await_resource_deleted(NFO_DOCUMENT, file_inside_victim_id)
Ivan Frade's avatar
Ivan Frade committed
286

287 288 289 290
        result = self.__get_text_documents()
        self.assertEqual(len(result), 1)
        unpacked_result = [r[0] for r in result]
        self.assertIn(self.uri("test-monitored/file1.txt"), unpacked_result)
Ivan Frade's avatar
Ivan Frade committed
291 292

        # Restore the dirs
293 294
        os.makedirs(self.path("test-monitored/dir1"))
        os.makedirs(self.path("test-monitored/dir1/dir2"))
Ivan Frade's avatar
Ivan Frade committed
295 296
        for f in ["test-monitored/dir1/file2.txt",
                  "test-monitored/dir1/dir2/file3.txt"]:
297
            filename = self.path(f)
298 299 300 301
            writer = open(filename, "w")
            writer.write("Don't panic, everything is fine")
            writer.close()
            self.system.store.await_resource_inserted(NFO_DOCUMENT, self.uri(f))
Ivan Frade's avatar
Ivan Frade committed
302 303

        # Check everything is fine
304 305 306
        result = self.__get_text_documents()
        self.assertEqual(len(result), 3)

Ivan Frade's avatar
Ivan Frade committed
307 308

if __name__ == "__main__":
309
    ut.main(failfast=True)