path: root/tests/functional-tests/miner/test_basic.py

# 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

from miner_testcase import MinerTestCase, MINER_TMP_DIR, uri, path


class MinerCrawlTest (MinerTestCase):

    """
    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 ;
                 nie:url ?url.
          }
          """)

    def __get_parent_urn(self, filepath):
        result = self.tracker.query ("""
          SELECT nfo:belongsToContainer(?u) WHERE {
              ?u a nfo:FileDataObject ;
                 nie:url \"%s\" .
          }
          """ % (uri (filepath)))
        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\" .
          }
          """ % (uri (filepath)))
        self.assertEquals(len(result), 1)
        return result[0][0]

    """
    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
        """
        # Maybe the information hasn't been committed yet
        time.sleep(1)
        result = self.__get_text_documents()
        self.assertEquals(len(result), 3)
        unpacked_result = [r[0] for r in result]
        self.assertIn(uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(uri("test-monitored/dir1/file2.txt"), unpacked_result)
        self.assertIn(
            uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)

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

    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
        """
        source = os.path.join(MINER_TMP_DIR, "test-no-monitored", "file0.txt")
        dest = os.path.join(MINER_TMP_DIR, "test-monitored", "file0.txt")
        shutil.copyfile(source, dest)

        dest_id, dest_urn = self.system.store.await_resource_inserted(
            'nfo:TextDocument', uri(dest))

        # verify if miner indexed this file.
        result = self.__get_text_documents()
        self.assertEquals(len(result), 4)
        unpacked_result = [r[0] for r in result]
        self.assertIn(uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(uri("test-monitored/dir1/file2.txt"), unpacked_result)
        self.assertIn(
            uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)
        self.assertIn(uri("test-monitored/file0.txt"), unpacked_result)

        # Clean the new file so the test directory is as before
        log("Remove and wait")
        os.remove(dest)
        self.system.store.await_resource_deleted(dest_id)

    def test_03_copy_from_monitored_to_unmonitored(self):
        """
        Copy an file from a monitored location to an unmonitored location
        Nothing should change
        """

        # Copy from monitored to unmonitored
        source = os.path.join(MINER_TMP_DIR, "test-monitored", "file1.txt")
        dest = os.path.join(MINER_TMP_DIR, "test-no-monitored", "file1.txt")
        shutil.copyfile(source, dest)

        time.sleep(1)
        # Nothing changed
        result = self.__get_text_documents()
        self.assertEquals(len(result), 3, "Results:" + str(result))
        unpacked_result = [r[0] for r in result]
        self.assertIn(uri("test-monitored/file1.txt"), unpacked_result)
        self.assertIn(uri("test-monitored/dir1/file2.txt"), unpacked_result)
        self.assertIn(
            uri("test-monitored/dir1/dir2/file3.txt"), unpacked_result)

        # Clean the file
        os.remove(dest)

    def test_04_copy_from_monitored_to_monitored(self):
        """
        Copy a file between monitored directories
        """
        source = os.path.join(MINER_TMP_DIR, "test-monitored", "file1.txt")
        dest = os.path.join(
            MINER_TMP_DIR, "test-monitored", "dir1", "dir2", "file-test04.txt")
        shutil.copyfile(source, dest)

        dest_id, dest_urn = self.system.store.await_resource_inserted(
            'nfo:TextDocument', uri(dest))

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

        # Clean the file
        os.remove(dest)
        self.system.store.await_resource_deleted(dest_id)
        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
        """
        source = os.path.join(MINER_TMP_DIR, "test-no-monitored", "file0.txt")
        dest = os.path.join(
            MINER_TMP_DIR, "test-monitored", "dir1", "file-test05.txt")
        shutil.move(source, dest)
        dest_id, dest_urn = self.system.store.await_resource_inserted(
            'nfo:TextDocument', uri(dest))

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

        # Clean the file
        os.remove(dest)
        self.system.store.await_resource_deleted(dest_id)
        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
        """
        source = path("test-monitored/dir1/file2.txt")
        dest = path("test-no-monitored/file2.txt")
        source_id = self.system.store.get_resource_id(uri(source))
        shutil.move(source, dest)
        self.system.store.await_resource_deleted(source_id)

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

        # Restore the file
        shutil.move(dest, source)
        self.system.store.await_resource_inserted(
            'nfo:TextDocument', uri(source))
        self.assertEquals(
            3, self.tracker.count_instances("nfo:TextDocument"))

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

        source = path("test-monitored/dir1/file2.txt")
        dest = path("test-monitored/file2.txt")

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

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

        shutil.move(source, dest)
        self.tracker.await_property_changed(resource_id, 'nie:url')

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

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

        # Restore the file
        shutil.move(dest, source)
        self.tracker.await_property_changed(resource_id, 'nie:url')

        result = self.__get_text_documents()
        self.assertEquals(len(result), 3)
        unpacked_result = [r[0] for r in result]
        self.assertIn(uri("test-monitored/dir1/file2.txt"), unpacked_result)

    def test_08_deletion_single_file(self):
        """
        Delete one of the files
        """
        victim = path("test-monitored/dir1/file2.txt")
        victim_id = self.system.store.get_resource_id(uri(victim))
        os.remove(victim)
        self.system.store.await_resource_deleted(victim_id)

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

        # Restore the file
        f = open(victim, "w")
        f.write("Don't panic, everything is fine")
        f.close()
        self.system.store.await_resource_inserted(
            'nfo:TextDocument', uri(victim))

    def test_09_deletion_directory(self):
        """
        Delete a directory
        """
        victim = path("test-monitored/dir1")
        victim_id = self.system.store.get_resource_id(uri(victim))
        shutil.rmtree(victim)

        file_inside_victim_url = 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(file_inside_victim_id)

        result = self.__get_text_documents()
        self.assertEquals(len(result), 1)
        unpacked_result = [r[0] for r in result]
        self.assertIn(uri("test-monitored/file1.txt"), unpacked_result)

        # Restore the dirs
        os.makedirs(path("test-monitored/dir1"))
        os.makedirs(path("test-monitored/dir1/dir2"))
        for f in ["test-monitored/dir1/file2.txt",
                  "test-monitored/dir1/dir2/file3.txt"]:
            filename = path(f)
            writer = open(filename, "w")
            writer.write("Don't panic, everything is fine")
            writer.close()
            self.system.store.await_resource_inserted(
                'nfo:TextDocument', uri(f))

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