TestTransfer.py #2

# This test needs to
# Use two servers identified by rsh hack P4PORT entries
# Create a stream hierarchy in one server
# Use the script to transfer the data to the target server
# Verify the outcome

# Tests can be done by invoking Python directly with the script (to test arguments)
# and to import the module and call the class directly (avoiding double invocation of Python)

from __future__ import print_function
import unittest
import os
import time
import shutil
import P4
from TransferStreamHierarchy import StreamHierarchy


def on_rm_tree_error(function, path, exc_info):
    os.chmod(path, os.stat.S_IWRITE)
    os.remove(path)


def ensure_directory(directory):
    if not os.path.isdir(directory):
        os.makedirs(directory, exist_ok=True)


class TestTransfer(unittest.TestCase):

    def setUp(self):
        self.start_dir = os.getcwd()

        self.source_root = os.path.join(self.start_dir, 'source')
        self.target_root = os.path.join(self.start_dir, 'target')

        self.cleanupTestTree(self.source_root)
        self.cleanupTestTree(self.target_root)

        ensure_directory(self.source_root)
        ensure_directory(self.target_root)

        self.p4d = "p4d"

        self.source_port = "rsh:{} -r \"{}\" -L log -vserver=3 -i".format( self.p4d, self.source_root )
        self.target_port = "rsh:{} -r \"{}\" -L log -vserver=3 -i".format(self.p4d, self.target_root)

        self.source_p4 = P4.P4(port=self.source_port)
        self.target_p4 = P4.P4(port=self.target_port)

        with self.source_p4.connect() as ps, self.target_p4.connect() as pt:
            depot = ps.fetch_depot('-t', 'stream', 'stream')
            ps.save_depot(depot)
            pt.save_depot(depot)

    def tearDown(self):
        if self.source_p4.connected():
            self.source_p4.disconnect()
        if self.target_p4.connected():
            self.target_p4.disconnect()

        time.sleep(1)

        self.cleanupTestTree(self.source_root)
        self.cleanupTestTree(self.target_root)

    def cleanupTestTree(self, server_root):
        os.chdir(self.start_dir)
        if os.path.isdir(server_root):
            shutil.rmtree(server_root, False, on_rm_tree_error)

    def testConnection(self):
        with self.source_p4.connect() as ps, self.target_p4.connect() as pt:

            source_info = ps.run_info()
            target_info = pt.run_info()

            self.assertEqual(len(source_info), 1, "Source cannot connect")
            self.assertEqual(len(target_info), 1, "Target cannot connect")

            source_depots = ps.run_depots()
            target_depots = pt.run_depots()

            self.assertEqual(len(source_depots), 2, "Cannot find 2 depots in source")
            self.assertEqual(len(target_depots), 2, "Cannot find 2 depots in target")

    def testMainStream(self):
        main_stream_name = '//stream/main'

        with self.source_p4.connect() as ps:
            main_stream = ps.fetch_stream('-t', 'mainline', main_stream_name)
            ps.save_stream(main_stream)

        # now transfer the stuff
        self.transferStreams(main_stream_name)

        with self.target_p4.connect() as pt:
            streams = pt.run_streams()

            self.assertEqual(len(streams), 1, "Did not find the stream in the target")
            self.assertEqual(streams[0]["Stream"], main_stream_name, "Stream names do not match")

    def testSimpleHierarchy(self):
        main_stream_name = '//stream/main'
        dev_stream_name = '//stream/dev'
        rel_stream_name = '//stream/rel'

        with self.source_p4.connect() as ps:
            main_stream = ps.fetch_stream('-t', 'mainline', main_stream_name)
            ps.save_stream(main_stream)
            dev_stream = ps.fetch_stream('-t', 'development', '-P', main_stream_name, dev_stream_name)
            ps.save_stream(dev_stream)
            rel_stream = ps.fetch_stream('-t', 'release', '-P', main_stream_name, rel_stream_name)
            ps.save_stream(rel_stream)

        # now transfer the stuff
        self.transferStreams(main_stream_name)

        with self.target_p4.connect() as pt:
            streams = pt.run_streams()

            self.assertEqual(len(streams), 3, "Did not find all streams in the target")

    def testIndirectReleaseHierarchy(self):
        main_stream_name = '//stream/main'
        rel_stream_name = '//stream/rel'
        rel1_stream_name = '//stream/rel1'

        with self.source_p4.connect() as ps:
            main_stream = ps.fetch_stream('-t', 'mainline', main_stream_name)
            ps.save_stream(main_stream)
            rel_stream = ps.fetch_stream('-t', 'release', '-P', main_stream_name, rel_stream_name)
            ps.save_stream(rel_stream)
            rel1_stream = ps.fetch_stream('-t', 'release', '-P', rel_stream_name, rel1_stream_name)
            ps.save_stream(rel1_stream)

        # now transfer the stuff
        self.transferStreams(main_stream_name)

        with self.target_p4.connect() as pt:
            streams = pt.run_streams()

            self.assertEqual(len(streams), 3, "Did not find all streams in the target")

    def testIndirectDevelopmentHierarchy(self):
        main_stream_name = '//stream/main'
        dev_stream_name = '//stream/dev'
        dev1_stream_name = '//stream/dev1'

        with self.source_p4.connect() as ps:
            main_stream = ps.fetch_stream('-t', 'mainline', main_stream_name)
            ps.save_stream(main_stream)
            rel_stream = ps.fetch_stream('-t', 'development', '-P', main_stream_name, dev_stream_name)
            ps.save_stream(rel_stream)
            rel1_stream = ps.fetch_stream('-t', 'development', '-P', dev_stream_name, dev1_stream_name)
            ps.save_stream(rel1_stream)

        # now transfer the stuff
        self.transferStreams(main_stream_name)

        with self.target_p4.connect() as pt:
            streams = pt.run_streams()

            self.assertEqual(len(streams), 3, "Did not find all streams in the target")

    def testVirtualHierarchy(self):
        main_stream_name = '//stream/main'
        virtual_stream_name = '//stream/virtual'
        devel_stream_name = '//stream/dev'

        with self.source_p4.connect() as ps:
            main_stream = ps.fetch_stream('-t', 'mainline', main_stream_name)
            ps.save_stream(main_stream)
            rel_stream = ps.fetch_stream('-t', 'virtual', '-P', main_stream_name, virtual_stream_name)
            ps.save_stream(rel_stream)
            rel1_stream = ps.fetch_stream('-t', 'development', '-P', virtual_stream_name, devel_stream_name)
            ps.save_stream(rel1_stream)

        # now transfer the stuff
        self.transferStreams(main_stream_name)

        with self.target_p4.connect() as pt:
            streams = pt.run_streams()

            self.assertEqual(len(streams), 3, "Did not find all streams in the target")

    def testRename(self):
        main_stream_name = '//stream/main'
        rel_stream_name = '//stream/rel'
        rel1_stream_name = '//stream/rel1'
        dev_stream_name = '//stream/dev'
        dev1_stream_name = '//stream/dev1'
        virtual_stream_name = '//stream/virtual'

        with self.source_p4.connect() as ps:
            main_stream = ps.fetch_stream('-t', 'mainline', main_stream_name)
            ps.save_stream(main_stream)
            rel_stream = ps.fetch_stream('-t', 'development', '-P', main_stream_name, dev_stream_name)
            ps.save_stream(rel_stream)
            rel1_stream = ps.fetch_stream('-t', 'development', '-P', dev_stream_name, dev1_stream_name)
            ps.save_stream(rel1_stream)
            rel_stream = ps.fetch_stream('-t', 'release', '-P', main_stream_name, rel_stream_name)
            ps.save_stream(rel_stream)
            rel1_stream = ps.fetch_stream('-t', 'release', '-P', rel_stream_name, rel1_stream_name)
            ps.save_stream(rel1_stream)
            virtual_stream = ps.fetch_stream('-t', 'virtual', '-P', main_stream_name, virtual_stream_name)
            ps.save_stream(virtual_stream)

        # create the new target depot with different stream depth

        with self.target_p4.connect() as pt:
            newdepot = pt.fetch_depot('-t', 'stream', 'newstream')
            newdepot._streamdepth = '//newstream/1/2'
            pt.save_depot(newdepot)

        # now transfer the stuff
        new_location = '//newstream/project/'
        self.transferStreams(main_stream_name, rename=('//stream/',new_location))

        with self.target_p4.connect() as pt:
            streams = pt.run_streams()

            self.assertEqual(len(streams), 6, "Did not find all streams in the target")
            stream1 = streams[0]
            self.assertTrue(stream1["Stream"].startswith(new_location), "Stream not at expected location")


    def transferStreams(self, main_stream, rename = ()):
        args = dict()
        args["port"] = self.source_port
        args["target"] = self.target_port
        args["user"] = ""
        args["target_user"] = ""
        args["rename"] = rename

        hierarchy = StreamHierarchy(**args)
        hierarchy.retrieve_hierarchy(main_stream)
        hierarchy.transfer_streams()

if __name__ == '__main__':
    unittest.main()