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

# 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

Ivan Frade's avatar
Ivan Frade committed
38 39 40 41 42 43 44 45
class MinerCrawlTest (CommonTrackerMinerTest):
    """
    Test cases to check if miner is able to monitor files that are created, deleted or moved
    """
    def __get_text_documents (self):
        return self.tracker.query ("""
          SELECT ?url WHERE {
              ?u a nfo:TextDocument ;
46
                 nie:url ?url.
Ivan Frade's avatar
Ivan Frade committed
47 48 49
          }
          """)

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

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

Ivan Frade's avatar
Ivan Frade committed
70 71 72 73 74 75 76 77 78 79
    """
    Boot the miner with the correct configuration and check everything is fine
    """
    def test_01_initial_crawling (self):
        """
        The precreated files and folders should be there
        """
        result = self.__get_text_documents ()
        self.assertEquals (len (result), 3)
        unpacked_result = [ r[0] for r in result]
80 81 82
        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)
83

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


## 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
90 91
##       tests to different classes

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

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

103
        # verify if miner indexed this file.
Ivan Frade's avatar
Ivan Frade committed
104 105 106
        result = self.__get_text_documents ()
        self.assertEquals (len (result), 4)
        unpacked_result = [ r[0] for r in result]
107 108 109 110
        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
111 112

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

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

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

        time.sleep (1)
        # Nothing changed
        result = self.__get_text_documents ()
131
        self.assertEquals (len (result), 3, "Results:" + str(result))
Ivan Frade's avatar
Ivan Frade committed
132
        unpacked_result = [ r[0] for r in result]
133 134 135
        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)
136

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

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

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

        result = self.__get_text_documents ()
        self.assertEquals (len (result), 4)
        unpacked_result = [ r[0] for r in result]
153 154 155 156
        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)
157

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


    def test_05_move_from_unmonitored_to_monitored (self):
        """
        Move a file from unmonitored to monitored directory
        """
168 169
        source = os.path.join (self.workdir, "test-no-monitored", "file0.txt")
        dest = os.path.join (self.workdir, "test-monitored", "dir1", "file-test05.txt")
Ivan Frade's avatar
Ivan Frade committed
170
        shutil.move (source, dest)
171
        dest_id, dest_urn = self.system.store.await_resource_inserted (NFO_DOCUMENT, self.uri(dest))
Ivan Frade's avatar
Ivan Frade committed
172 173 174 175

        result = self.__get_text_documents ()
        self.assertEquals (len (result), 4)
        unpacked_result = [ r[0] for r in result]
176 177 178 179
        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)
180

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

## """ move operation and tracker-miner response test cases """
## class move(TestUpdate):


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

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

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


    def test_07_move_from_monitored_to_monitored (self):
        """
        Move a file between monitored directories
        """
216

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

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

        source_dir_urn = self.__get_file_urn (os.path.dirname(source))
223 224 225
        parent_before = self.__get_parent_urn (source)
        self.assertEquals (source_dir_urn, parent_before)

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

229 230
        # Checking fix for NB#214413: After a move operation, nfo:belongsToContainer
        # should be changed to the new one
231
        dest_dir_urn = self.__get_file_urn (os.path.dirname(dest))
232 233 234 235
        parent_after = self.__get_parent_urn (dest)
        self.assertNotEquals (parent_before, parent_after)
        self.assertEquals (dest_dir_urn, parent_after)

Ivan Frade's avatar
Ivan Frade committed
236 237 238
        result = self.__get_text_documents ()
        self.assertEquals (len (result), 3)
        unpacked_result = [ r[0] for r in result]
239 240 241
        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
242 243 244

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

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


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

        result = self.__get_text_documents ()
        self.assertEquals (len (result), 2)
        unpacked_result = [ r[0] for r in result]
265 266
        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
267 268 269 270 271

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

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

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

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

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

        # Check everything is fine
        result = self.__get_text_documents ()
        self.assertEquals (len (result), 3)

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