# We install GhettoCI directly under our UNIX user home folder
cd ~
virtualenv -p python3 vvv-venv
source vvv-venv/bin/activate
pip install vvv

# ghetto-ci now lives in vvv-venv/bin/ghetto-ci

# Running the script, using the Python environment prepared
# to see that everything works (source command above
# has added this to PATH)
ghetto-ci -h

# Will print output to console because email notification details are not given
ghetto-ci /my/svn/repo /tmp/status-file.ci "cd /my/svn/repo && bin/test"

#/bin/sh
sudo -i -u yourunixuser "source vvv-venv/bin/activate && \
  ghetto-ci /my/svn/repo /tmp/status-file.ci 'cd /my/svn/repo && bin/test'"

ghetto-ci -force -alwaysoutput ...email settings here... /repo /tmp/status-file.ci true

#!/bin/sh
#
# Example ghetto-ci integration for Plone buildout and custom add-ons.
# Using dummy gmail account for outgoing messages.
#
# Install VVV under buildout
#
# Run this file in buildout main folder:
#
# cd ~/mybuildoutfolder
# src/my-repo/continuos-integration.sh
#
#

# The list of persons who will receive resports of test status changes
RECEIVERS="person@asdf.com, person2@asdf.com, person3@example.com"

# Ghetto-CI template command which is run against multiple repos / multiple test commands
# We use localhost 25 as the SMTP server -> assume your UNIX server has postfix
# or something configured... could be gmail.com servers also here
GHETTOCI="vvv-venv/bin/ghetto-ci \
    -smtpserver smtp.gmail.com \
    -smtpport 465 \
    -smtpuser mailer@gmail.com \
    -smtppassword OMGITISFULLofKITT3NS \
    -receivers \"$RECEIVERS\" \
    -envelopefrom \"Continuous integration service <mailer@gmail.com>\" \
    -smtpfrom mailer@gmail.com"

# Note that SVN revision info is folding down in the folders
# so you can target tests to a specific SVN repo subfolder

# Note: eval needs to be used due to shell script quotation mark fuzz

# See that buildout completes (no changes in the external environment, like someone accidentally
# publishing broken packages on PyPI). We actually place buildout.cfg under this SVN repo
# and then just symlink it
eval $GHETTOCI src/my-repo buildout.ci 'bin/buildout'

# Run tests against hospital product
eval $GHETTOCI src/my-repo/Products.Hospital hospital.ci \"bin/test -s Products.Hospital\"

# Run tests against patient product
eval $GHETTOCI src/my-repo/Products.Patient patient.ci \"bin/test -s Products.Patient\"

__author__ = "Mikko Ohtamaa <mikko@opensourcehacker.com>"
__license__ = "WTFPL"

# pylint complains about abstract Python 3 members and ABCMeta, this will be fixed in the future
# pylint: disable=R0201, W0611

# Python imports
from abc import ABCMeta, abstractmethod
from email.mime.text import MIMEText
import pickle
import os
import sys
import subprocess
from smtplib import SMTP_SSL, SMTP 

#: Template used in email notifications
NOTIFICATION_BODY_TEMPLATE = """
Last commit %(commit)s by %(author)s:

%(commit_message)s

Test output:

%(test_output)s
"""

def split_first(line, separator):
    """
    Split a string to (first part, remainder)
    """
    parts = line.split(separator)
    return parts[0], separator.join(parts[1:])

def shell(cmdline):
    """
    Execute a shell command.

    :returns: (exitcode, stdout / stderr output as string) tuple
    """

    process = subprocess.Popen(cmdline, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)

    # XXX: Support stderr interleaving
    out, err = process.communicate()

    # :E1103: *%s %r has no %r member (but some types could not be inferred)*
    # pylint: disable=E1103
    out = out.decode("utf-8")
    err = err.decode("utf-8")

    return (process.returncode, out + err)    

class Status:
    """
    Stored pickled CI status of a test run.

    Use Python pickling serialization for making status info persistent.
    """

    def __init__(self):

        #: Set to True of False by automation,
        # but we have a special value for the first run
        # to get output always
        self.test_success = "xxx"
        self.last_commit_id = None

    @classmethod
    def read(cls, path):
        """
        Read status file.

        Return fresh status if file does not exist.
        """    

        if not os.path.exists(path):
            # Status file do not exist, get default status
            return Status()

        f = open(path, "rb")

        try:
            return pickle.load(f)
        finally:
            f.close()

    @classmethod
    def write(cls, path, status):
        """
        Write status file
        """        
        f = open(path, "wb")
        pickle.dump(status, f)
        f.close()

class Repo(metaclass=ABCMeta):
    """
    Define interface for presenting one monitored software repository in ghetto-ci.
    """
    def __init__(self, path):
        """
        :param path: Abs FS path to monitored repository
        """
        self.path = path

    @abstractmethod
    def update(self):
        """ Update repo from version control """
        pass

    @abstractmethod
    def get_last_commit_info(self):
        """
        Get the last commit status.

        :return tuple(commit id, commiter, message, raw_output) or (None, None, None, raw_output) on error
        """
        return (None, None, None)

class SVNRepo(Repo):
    """ Handle Subversion repository update and last commit info extraction """

    def update(self):
        """
        Run repo update in a source folder
        """
        exitcode, output = shell("svn up %s" % self.path)
        return exitcode == 0, output

    def get_last_commit_info(self):
        """
        Get the last commit status.
        """

        # Get output from svn info
        info, output = self.get_svn_info()

        if not info:
            return (None, None, None, output)

        # Get output from svn log
        log_success, author, log = self.get_svn_log()
        if not log_success:
            return (None, None, None, log)

        return (info["Last Changed Rev"], author, log, output)

    def get_svn_log(self):
        """
        Extract the last commit author and message.

        :return: tuple (success, last commiter, output / last commit message)
        """
        exit_code, output = shell("svn log -l 1 %s" % self.path)

        if exit_code != 0:
            return (False, None, output)

        # ------------------------------------------------------------------------
        # r6101 | xxx | 2012-04-28 15:57:14 +0300 (Sat, 28 Apr 2012) | 1 line

        lines = output.split("\n")
        author_line = lines[1]
        author = author_line.split("|")[1].strip()

        return  (True, author, output)

    def get_svn_info(self):
        """
        Get svn info output parsed to dict
        """
        exit_code, output = shell("svn info %s" % self.path)
        if exit_code != 0:
            return None, output

        data = {}
        for line in output.split("\n"):
            key, value = split_first(line, ":")
            data[key] = value

        return data, output

class Notifier:
    """
    Intelligent spam being. Print messages to stdout and send emai if
    SMTP server details are available.
    """

    def __init__(self, server, port, username, password, receivers, from_address, envelope_from):
        """
        :param server: SMTP server

        :param port: SMTP port, autodetects SSL

        :param username: SMTP credentials 

        :param password: SMTP password

        :param receivers: String, comma separated list of receivers

        :param from_email: Sender's email address
        """

        self.server = server
        self.port = port
        self.username = username
        self.password = password
        self.receivers = receivers
        self.from_address = from_address
        self.envelope_from = envelope_from

    def send_email_notification(self, subject, output):
        """

        Further info:

        * http://docs.python.org/py3k/library/email-examples.html

        :param receivers: list of email addresses

        :param subject: Email subject

        :param output: Email payload
        """

        if not self.receivers:
            raise RuntimeError("Cannot send email - no receivers given")

        if self.port == 465:
            # SSL encryption from the start
            smtp = SMTP_SSL(self.server, self.port)
        else:
            # Plain-text SMTP, or opt-in to SSL using starttls() command
            smtp = SMTP(self.server, self.port)

        msg = MIMEText(output, "text/plain")
        msg['Subject'] = subject

        if self.envelope_from:
            msg['From'] = self.envelope_from
        else:
            msg['From'] = self.from_address

        # Add visible receivers header
        msg["To"] = self.receivers

        # SMTP never works on the first attempt...
        smtp.set_debuglevel(False)

        # SMTP authentication is optional
        if self.username and self.password:
            smtp.login(self.username, self.password)

        # Convert comma-separated sl
        receivers = [email.strip() for email in self.receivers.split(",")]

        try:
            smtp.sendmail(self.from_address, receivers, msg.as_string())
        finally:
            smtp.close()

    def print_notification(self, subject, output):
        """
        Dump the notification to stdout
        """
        print(subject)
        print("-" * len(subject))
        print(output) 

    def notify(self, subject, output):
        """
        Notify about the tests status.
        """
        if self.server:
            self.send_email_notification(subject, output)

        self.print_notification(subject, output)

def run_tests(test_command):
    """
    Run testing command.

    Assume exit code = 0 -> test success
    """
    exitcode, output = shell(test_command)
    return (exitcode == 0, output)

# Parsing command line with plac rocks
# http://plac.googlecode.com/hg/doc/plac.html
def main(smtpserver : ("SMTP server address for mail out. Required if you indent to use email notifications.", "option"),
         smtpport : ("SMTP server port for mail out", "option", None, int),
         smtpuser : ("SMTP server username", "option"),
         smtppassword : ("SMTP server password", "option"),
         smtpfrom : ("Notification email From address", "option"),
         envelopefrom : ("Verbose Name <from@site.com> sender address in outgoing email", "option"),
         receivers : ("Notification email receives as comma separated string", "option"),
         force : ("Run tests regardless if there have been any repository updates", "flag"),
         alwaysoutput : ("Print test run output regardless whether test status has changed since the last run", "flag"),

         repository : ("Monitored source control repository (SVN)", "positional"),
         statusfile : ("Status file to hold CI history of tests", "positional"),
         testcommand : ("Command to run tests. Exit code 0 indicates test success", "positional"),
         ):
    """
    Ghetto-CI

    A simple continuous integration server.

    Ghetto-CI will monitor the software repository.
    Give a (Subversion) software repository and a test command run test against it.
    Make this command run regularly e.g. using UNIX cron service.    
    You will get email notification when test command status changes from exit code 0.

    For more information see http://pypi.python.org/pypi/vvv
    """

    notifier = Notifier(server=smtpserver, port=smtpport,
                      username=smtpuser, password=smtppassword, from_address=smtpfrom,
                      receivers=receivers, envelope_from=envelopefrom)

    repo = SVNRepo(repository)    
    status = Status.read(statusfile)

    success, output = repo.update()

    # Handle repo update failure
    if not success:
        notifier.notify("Could not update repository: %s. Probably not valid SVN repo?\n" % repository, output)
        return 1

    commit_id, commit_author, commit_message, output = repo.get_last_commit_info()

    # Handle repo info failure
    if not commit_id:
        notifier.notify("Could not get commit info: %s\n%s" % (repository, output), "Check svn info by hand")
        return 1

    # See if repository status has changed
    if commit_id  != status.last_commit_id or force:
        test_success, output = run_tests(testcommand)
    else:
        # No new commits, nothing to do
        print("No changes in repo %s" % repository)
        return 0

    # Test run status have changed since last run
    if (test_success != status.test_success) or alwaysoutput:

        notification_body = NOTIFICATION_BODY_TEMPLATE % dict(commit = commit_id, author=commit_author,
            commit_message=commit_message, test_output=output)

        if test_success:
            subject = "Test now succeed for %s" % testcommand
        else:
            subject = "Test now fail for %s" % testcommand

        notifier.notify(subject, notification_body)

    # Update persistent test status
    new_status = Status()
    new_status.last_commit_id = commit_id
    new_status.test_success = test_success
    Status.write(statusfile, new_status)

    if test_success:
        return 0
    else:
        return 1

def entry_point():
    """
    Enter the via setup.py entry_point declaration.

    Handle UNIX style application exit values
    """
    import plac
    exitcode = plac.call(main)
    sys.exit(exitcode)

if __name__ == "__main__":
    entry_point()

 Subscribe to RSS feed Follow me on Twitter Follow me on Facebook Follow me Google+