VSS Integration Module (Python recipe) by Tim Watson

ActiveState Code (http://code.activestate.com/recipes/473884/)

#!/usr/bin/env python
"""
    Visual Source Safe Integration Module.
    
    Notes:
    
        Early stages of this module - GLV, check in and check out only. Some resource management stuff, but not
        iron-tight, so would recommend calling dispose() explicitly on all objects!
    
    Known issues:
    
    *   Doesn't seem to like UNC paths on my two boxes (+ lapdog); would recommend mapping a network drive and pointing to that.
    
"""
__author__  = "Tim Watson"
__version__ = "$Revision: 1.1 $"
__license__ = "Python"

import win32com.client as com;
import pywintypes as win32;

#internal constants
VSS_ROOT_FOLDER         = "$/";
#vs constants
VSSFILE_CHECKEDOUT      =0x1        # from enum VSSFileStatus
VSSFILE_CHECKEDOUT_ME   =0x2        # from enum VSSFileStatus
VSSFILE_NOTCHECKEDOUT   =0x0        # from enum VSSFileStatus
VSSITEM_FILE            =0x1        # from enum VSSItemType
VSSITEM_PROJECT         =0x0        # from enum VSSItemType
VSSFLAG_RECURSNO        =0x1000     # from enum VSSFlags
VSSFLAG_RECURSYES       =0x2000     # from enum VSSFlags

class ObjectDisposedException(Exception): pass
class ComException(Exception, win32.com_error): pass

class Disposable(object):
    def __init__(self):
        super(Disposable, self).__init__();
    
    def __del__(self):
        self.dispose();
    
    def dispose(self): pass
    
    def guard(self):
        if self.disposed == True: raise ObjectDisposedException("Object disposed!");

class Searchable:
    """Only works in conjunction with classes that implement the Node interface."""
    def __init__(self):
        super(Searchable, self).__init__();
        
    def __contains__(self, item):
        return self.contains(item);
    
    def contains(self, item):
        foundItem = None;
        try:
            foundItem = self.__getitem__(item);
            if foundItem:
                return True;
            else:
                return False;
        finally:
            foundItem.dispose();
    
    def __getitem__(self, key):
        item = None;
        if hasattr(self.vssItem, "Items"): 
            try:
                #do a quick search of the current branch
                item = self.vssItem.Items.Item(key);
            except win32.com_error:
                #try a recursive search for the item
                searches = [];
                try:
                    for child in self.descendants():
                        if child.name == key:
                            return Node(child);
                        searches.append(child);
                finally:
                    for n in searches: n.dispose();
        return Node(item);

class Node(Disposable, Searchable):
    def __init__(self, comobject):
        self.vssItem = comobject;
    
    def __call__(self):
        return self.__repr__();

    #are these really needed on subclasses?
    def __del__(self):
        self.dispose();

    def __iter__(self):
        return self.iterkeys();
    
    def __nonzero__(self):
        return self.vssItem is not None;
    
    def __repr__(self):
        return "<Node.%s>" % str(self.name);

    def iterkeys(self):
        """Gets the immediate descendants of the current node."""
        if self.vssItem is not None:
            if not hasattr(self.vssItem, "Items"): return;
            for item in self.vssItem.Items:
                yield Node(item);

    def descendants(self):
        """Iterates over all the descendants of the current node,
            in order, to the leaf level."""
        if self.vssItem is not None:
            for item in self.__gen(self.vssItem):
                yield Node(item);
    
    def __gen(self, root):
        yield root;
        if hasattr(root, "Items"):
            for node in root.Items:
                for item in self.__gen(node):
                    yield item;

    #todo: sort out this hideous duplication!
    def checkout(self, local=None, flags=None):
        if self.vssItem is not None:
            if local is not None:
                if flags is not None:
                    self.vssItem.Checkout(Local=local, iFlags=flags);
                else:
                    self.vssItem.Checkout(Local=local);
            else:
                if flags is not None:
                    self.vssItem.Checkout(iFlags=flags);
                else:
                    self.vssItem.Checkout();
    
    #todo: sort out this hideous duplication!
    def checkin(self, local=None, flags=None):
        if self.vssItem is not None:
            if local is not None:
                if flags is not None:
                    self.vssItem.Checkin(Local=local, iFlags=flags);
                else:
                    self.vssItem.Checkin(Local=local);
            else:
                if flags is not None:
                    self.vssItem.Checkin(iFlags=flags);
                else:
                    self.vssItem.Checkin();
    
    def get(self, local, flags=None):
        if local is None: return;
        if self.vssItem is not None:
            if flags is not None:
                self.vssItem.Get(Local=local, iFlags=flags);
            else:
                self.vssItem.Get(Local=local);
    
    def __process(self, function, local=None, flags=None):
        if function is None: return;
        if local is not None:
                if flags is not None:
                    function(Local=local, iFlags=flags);
                else:
                    function(iFlags=flags);
        else:
            if flags is not None:
                function(iFlags=flags);
            else:
                function();
    
    def __checkedOut(self):
        if not self.__isFile(): return False;
        return self.vssItem.IsCheckedOut == VSSFILE_CHECKEDOUT or self.vssItem.IsCheckedOut == VSSFILE_CHECKEDOUT_ME;
    
    def __checkedOutToMe(self):
        if self.__checkedOut():
            return self.vssItem.IsCheckedOut == VSSFILE_CHECKEDOUT_ME;
    
    def dispose(self):
        self.vssItem = None;

    def __getName(self):
        if self.vssItem is None:
            return "None";
        else:
            return self.vssItem.Name;

    def hasItems(self):
        return not hasattr(self.vssItem, "Items");
    
    def __isProject(self):
        return self.vssItem is not None and self.vssItem.Type == VSSITEM_PROJECT;
    
    def __isFile(self):
        return not self.__isProject();

    def __deleted(self):
        if self.vssItem is not None: 
            return self.vssItem.Deleted;
    
    def __parent(self):
        if self.vssItem is not None:
            try:
                return Node(self.vssItem.Parent);
            except: pass
        return Node(None);
    
    def __path(self):
        if self.vssItem is not None: return self.vssItem.Spec;
        return "None";

    #todo: figure out how to access self from a lambda!

    vsspath = property (fget=__path);

    deleted = property (
            fget=__deleted,
            doc="""Indicates whether or not the current instance has been deleted."""
        )

    name = property (
            fget=__getName,
            doc="""Returns the name of the underlying node object"""
        );
    
    parent = property (
            fget=__parent,
            doc="""Gets the parent for the current item."""
        );
    
    isCheckedOut = property (
            fget=__checkedOut,
            doc="""
                Indicates whether or not a file is checkout out.
                Returns false is the current item is a folder.
                """
        );
    
    isMine = property (
            fget=__checkedOutToMe,
            doc="""
                Indicates whether or not a file is checked out to the currently logged in user.
                """
        );
    
    isLeaf = property (
            fget=hasItems,
            doc="""Indicates whether or not this instance is at the leaf level."""
        );

    isFile = property (
            fget=__isFile,
            doc="""Indicates whether or not this instance represents a file."""
        );
    
    isProject = property (
            fget=__isProject,
            doc="""Indicates whether or not this instance represents a project."""
        );

class Database(Node):
    """ A vss database instance. """
    def __init__(self):
        "Database init";
        self.__vss = com.Dispatch("SourceSafe");
        super(Database, self).__init__(None);
        self.disposed = lambda: self.__vss is None;
        self.isOpen = lambda: self.vssItem is not None;
    
    def __del__(self):
        self.dispose();
    
    def __iter__(self):
        return self.iterkeys();
    
    def iterkeys(self):
        super(Database, self).guard();
        for item in self.__getRoot().Items:
            yield Node(item);

    def open(self, iniFile, username, password=None):
        """
            Opens a new instance of vss database.
        """
        super(Database, self).guard();
        self.__vss.Open(iniFile, username);
        self.vssItem = self.__vss.VSSItem(VSS_ROOT_FOLDER, False);
    
    @staticmethod
    def openProject(iniFile, projectName, username, password=None):
        """
            Opens and returns the project at the specified location in vss.
            If you pass an absolute path (e.g. a path including the vss separator character '/',
            will attempt to load the project from the exact path specified (adding the root namespace
            declaration '$/' if missing). If this attempt fails (or if you simply pass the name of the project,
            such as 'my project'), will create a new database at the root position and recursively search for the
            specified project.            
            
            Returns a new project node, or 'None', if the search fails. The recursive search is quite
            slow, so pass the full path if you can.
        """
        prName = projectName;
        if projectName.find("/") != -1:
            if not projectName.startswith(VSS_ROOT_FOLDER):
                prName = VSS_ROOT_FOLDER + projectName;
            try:
                vss = com.Dispatch("SourceSafe");
                vss.Open(iniFile, username);
                try:
                    p = vss.VSSItem(prName);
                    return Node(p);
                except: pass
            finally:
                vss = None;
            #out of luck - try stripping out the project name and doing a 'normal' search
            if projectName.endswith("/"):
                prName = projectName.rstrip("/");
            lastSepChar = prName.rfind("/");
            if lastSepChar == -1:
                raise Exception("I Can't be bothered parsing this mess :- put in a proper project name you blithering idiot!");
            prName = prName[lastSepChar, len(prName) - lastSepChar];
            if prName.startswith("/"):
                prName = prName.lstrip("/");
                #todo: all of the above with a nice, clean regex instead.
        ##regular search method:
        toplevels = [];
        db = None;
        try:
            try:
                db = Database();
                db.open(iniFile, username, password);
                #minimum depth search first:            
                for topLevelProj in db:
                    if topLevelProj.name == projectName:
                        return Node(topLevelProj);
                    toplevels.append(topLevelProj);
                #now try increasing the search depth exponentially:
                for proj in toplevels:
                    for child in proj.descendants():
                        if child.name == projectName: return Node(child);
                        else: child.dispose();
                #todo: do this (above) with a list comprehension instead!
            finally:
                db.dispose();
        finally:
            for p in toplevels:
                p.dispose();
    
    def dispose(self):
        """Disposes of the current instance."""
        self.__vss = None;
        super(Database, self).dispose();
   
    def hasItems(self):
        return True;
   
    def __getRoot(self):
        super(Database, self).guard();
        return self.vssItem;

class SourceSafe:
    """Simple vss wrapper, to make lifetime management easier."""
    def __init__(self, iniFile, username):
        self.iniFile = iniFile;
        self.username = username;
        self.checked_out = False;
        self.project = None;
        self.releaseFolder = None;
        self.verbose = False;
    
    def checkout(self, projectRoot, checkoutFolder, iflags=None):
        if self.verbose: print "Attempting to check out files...";
        if self.project is None:
            self.project = Database.openProject(str(self.iniFile), projectRoot, str(self.username));
        assert self.project is not None;
        if projectRoot == checkoutFolder:
            self.releaseFolder = self.project;
        else:
            self.releaseFolder = self.project[checkoutFolder];
        assert self.releaseFolder, "Failed to locate vss folder %s!" % checkoutFolder;
        try:
            self.releaseFolder.checkout(flags=iflags);
            self.checked_out = True;
            if self.verbose: print "Check out successful.";
        except:
            print "Check out failed...";

    def checkin(self, iflags=None):
        if self.checked_out:
            if self.verbose: print "Attempting to check in files...";
            try:
                self.releaseFolder.checkin(flags=iflags);
                if self.verbose: print "Check in successful.";
            except:
                print "Check in failed..";
    
    def get(self, project, local=None, iflags=None):
        try:
            if self.project is None:
                if project is None: return;
                self.project = Database.openProject(str(self.iniFile), project, str(self.username));
            if self.verbose: print "Attempting [Get Latest Version]...";
            self.releaseFolder = self.project;
            self.releaseFolder.get(local, iflags);
            if self.verbose: print "[Get Latest Version] complete...";
        except:
            print "[Get Latest Version] failed...";

    def dispose(self):
        if self.project is not None: self.project.dispose();
        self.project = None;
        if self.releaseFolder is not None: self.releaseFolder.dispose();
        self.releaseFolder = None;

###########################################################################

#!/usr/bin/env python
"""
    Unit test module for vss module.
"""
    
__author__  = "Tim Watson"
__version__ = "$Revision: 1.1 $"
__license__ = "Python"

import unittest;
from vss import *;

DATABASE    = "C:\\BUILD_TEST\\srcsafe.ini";
USER        = "scripting.client"
PROJECT     = "ICBFData";
PROJECT_PARTIAL = "BI_Build/" + PROJECT;
PROJECT_FULL    = "$/" + PROJECT_PARTIAL;

class ServerFixture(unittest.TestCase):
    
    def setUp(self):
        self.db = Database();
        self.db.open(DATABASE, USER);
    
    def tearDown(self):
        self.db.dispose();
    
    def testConnect(self):
        "Tests the ability to create a new server instance."
        self.db.dispose();
        self.assert_(self.db.disposed, "not disposed!");
    
    def testGetRootProjectAndIter(self):
        for project in self.db:
            self.failIf(project.name is None, "Project name cannot be None");
    
    def testRecursiveIterProjects(self):
        MAX_COUNT = 100;
        count = 0;
        for proj in self.db:
            for node in proj.descendants():
                if count > MAX_COUNT: return;
                count += 1;
                self.failIf(node.name is None);
                print node.name;
                if node.isLeaf:
                    self.failIf(hasattr(node, "Items"), "Node %s was supposed to be a leaf!" % node.name);
    
    def testContainmentCheckSucceeds(self):
        BAD_ITEM_NAME = "Doesn't Exist!";
        GOOD_ITEM_NAME = "IABuild";
        self.assertFalse(BAD_ITEM_NAME in self.db);
        self.failIf(not GOOD_ITEM_NAME in self.db);
        
    def testGetBadProjectYieldsNone(self):
        BAD_PROJECT = "Not Real";
        self.assertEqual(Database.openProject(DATABASE, BAD_PROJECT, USER), None);
      
    def testPartialProjectOk(self):
        self.doProjectLoad(PROJECT_PARTIAL);
    
    def testFullProjectOk(self):
        self.doProjectLoad(PROJECT_FULL);
    
    def doProjectLoad(self, project):
        proj = Database.openProject(DATABASE, project, USER);
        self.assertFalse(proj is None, "Failed to locate project");
        self.assertEqual(proj.name, PROJECT);

if __name__ == "__main__":
    unittest.main();