Skip to content

Instantly share code, notes, and snippets.

View harlowja's full-sized avatar

Joshua Harlow harlowja

  • Waymo
  • Campbell
  • 05:10 (UTC -07:00)
View GitHub Profile
@harlowja
harlowja / remote_tester
Created June 9, 2014 19:22
Remote tox tester
import argparse
import getpass
import os
import select
import socket
import sys
import tarfile
import tempfile
import threading
import eventlet
eventlet.monkey_patch(os=False, thread=False)
import time
import threading
LOCK = threading.Lock()
@harlowja
harlowja / gist:6a185742de6d4887efe4
Created June 10, 2014 20:12
Rpm version mini-webserver
#!/usr/bin/python
from BaseHTTPServer import (HTTPServer, BaseHTTPRequestHandler)
from optparse import OptionParser
from rpmUtils.miscutils import splitFilename
from yum import packages
import StringIO
import cgi
@harlowja
harlowja / rpm_compare.py
Created June 16, 2014 23:10
Rpm comparator test
from yum import packages
def to_envra(name, version, arch="noarch", epoch=0):
pieces = version.split(".")
version_segments = []
release_segments = []
in_release = True
for segment in version.split("."):
if segment.isdigit() and in_release:
@harlowja
harlowja / slow_queue_test.py
Created June 19, 2014 03:42
Slow queue workers
import time
import threading
import sloq
THREADS = 5
RATE = 1
WORK_UNITS = 100
WORK_DELAY = 0.1
@harlowja
harlowja / test.py
Created July 2, 2014 20:42
Storage idea
import abc
import six
@six.add_metaclass(abc.ABCMeta)
class BaseStorage(object):
def ensure_atom(self, atom):
def set_atom_state(self, atom_name, state):
import a
class B(a.A):
pass
def itersubclasses(cls, _seen=None):
"""Generator over all subclasses of a given class in depth first order."""
from taskflow.types import fsm
from taskflow import states as st
m = fsm.FSM(st.UNDEFINED)
m.add_state(st.UNDEFINED)
m.add_state(st.RESUMING)
m.add_state(st.REVERTED)
m.add_state(st.SCHEDULING)
@harlowja
harlowja / fsm_test.py
Created July 12, 2014 20:18
fsm-test
from taskflow.types import fsm
from taskflow import states as st
m = fsm.FSM(st.UNDEFINED)
def on_enter(new_state, event):
print("Entering new state '%s' in "
"response to event '%s'" % (new_state, event))
# -*- coding: utf-8 -*-
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#