Python模拟进程用于单元测试

8

背景:
我正在使用Python编写一个进程监控工具(Windows和Linux),并实现单元测试覆盖率。该进程监视器钩入Windows API功能EnumProcesses在Windows上,并监视/proc目录以查找当前进程。然后将进程名称和进程ID写入日志,可供单元测试使用。

问题:
当我对监视行为进行单元测试时,我需要启动和终止进程。如果有一种(跨平台的?)方法可以启动和终止虚假系统进程,我会很高兴,我可以给它起一个独特的名称(并在单元测试中跟踪其创建)。

初始想法:

  • 我可以使用subprocess.Popen()打开任何系统进程,但这会遇到一些问题。如果系统也运行我用于测试的进程,则单元测试可能会错误地通过。此外,单元测试是从命令行运行的,我能想到的任何Linux进程都会挂起终端(如nano等)。
  • 我可以通过其进程ID启动进程并跟踪它,但我不确定如何在不挂起终端的情况下执行此操作。

这些只是来自初始测试的想法和观察,如果有人能在这两个方面证明我错了,那将不胜感激。

我正在使用Python 2.6.6。

编辑:
获取所有Linux进程ID:

try:
    processDirectories = os.listdir(self.PROCESS_DIRECTORY)
except IOError:
    return []
return [pid for pid in processDirectories if pid.isdigit()]

获取所有Windows进程ID:

import ctypes, ctypes.wintypes

Psapi = ctypes.WinDLL('Psapi.dll')
EnumProcesses = self.Psapi.EnumProcesses
EnumProcesses.restype = ctypes.wintypes.BOOL

count = 50
while True:
    # Build arguments to EnumProcesses
    processIds = (ctypes.wintypes.DWORD*count)()
    size = ctypes.sizeof(processIds)
    bytes_returned = ctypes.wintypes.DWORD()
    # Call enum processes to find all processes
    if self.EnumProcesses(ctypes.byref(processIds), size, ctypes.byref(bytes_returned)):
        if bytes_returned.value &lt size:
            return processIds
       else:
            # We weren't able to get all the processes so double our size and try again
            count *= 2
    else:
        print "EnumProcesses failed"
        sys.exit()

Windows code is from here


只是说一下:你正在构建一个集成测试,而不是单元测试;单元测试只关注项目中的一个部分,其他所有内容都被存根和假定为正常工作。 - Martijn Pieters
不是回答你的问题,但是... 你知道psutil吗?它可能会处理你的应用程序中大部分无聊/困难的部分。 - dbn
2个回答

7

编辑:这个答案有点长 :)

你的代码和我的原始答案并没有太大区别。我的一些想法仍然适用。

在编写单元测试时,您希望仅测试您的逻辑。当您使用与操作系统交互的代码时,通常要模拟该部分。原因是您对这些库的输出控制不大,就像您发现的那样。因此,更容易模拟这些调用。

在这种情况下,有两个库正在与系统交互:os.listdirEnumProcesses。由于您没有编写它们,我们可以轻松地伪造它们以返回我们需要的内容。在这种情况下是一个列表。

但是,等等,在您的评论中提到:

"我遇到的问题是它确实没有测试 我的代码是否看到了系统上的新进程,而是代码是否正确监视了列表中的新项。"

事实是,我们不需要测试实际监视系统进程的代码,因为它是第三方代码。我们需要测试的是您的代码逻辑如何处理返回的进程。因为那是你写的代码。我们测试列表的原因是因为那是您的逻辑所做的。 os.listdirEniumProcesses返回一个pid列表(数字字符串和整数,分别),您的代码对该列表进行操作。

我假设您的代码在类中(您的代码中使用了self)。我还假设它们隔离在自己的方法中(您使用了return)。这将与我最初建议的有些类似,只是具有实际的代码:)我不知道它们是否在同一个类中或不同的类中,但这并不重要。

Linux方法

现在,测试您的Linux进程函数并不那么困难。您可以修补os.listdir以返回pid列表。

def getLinuxProcess(self):
    try:
        processDirectories = os.listdir(self.PROCESS_DIRECTORY)
    except IOError:
        return []
    return [pid for pid in processDirectories if pid.isdigit()]

现在开始测试。

import unittest
from fudge import patched_context
import os
import LinuxProcessClass # class that contains getLinuxProcess method

def test_LinuxProcess(self):
    """Test the logic of our getLinuxProcess.

       We patch os.listdir and return our own list, because os.listdir
       returns a list. We do this so that we can control the output 
       (we test *our* logic, not a built-in library's functionality).
    """

    # Test we can parse our pdis
    fakeProcessIds = ['1', '2', '3']
    with patched_context(os, 'listdir', lamba x: fakeProcessIds):
        myClass = LinuxProcessClass()
        ....
        result = myClass.getLinuxProcess()

        expected = [1, 2, 3]
        self.assertEqual(result, expected)

    # Test we can handle IOERROR
    with patched_context(os, 'listdir', lamba x: raise IOError):
        myClass = LinuxProcessClass()
        ....
        result = myClass.getLinuxProcess()

        expected = []
        self.assertEqual(result, expected)

    # Test we only get pids
    fakeProcessIds = ['1', '2', '3', 'do', 'not', 'parse']
    .....

Windows 方法

测试 Windows 方法有点棘手。我的建议如下:

def prepareWindowsObjects(self):
    """Create and set up objects needed to get the windows process"
    ...
    Psapi = ctypes.WinDLL('Psapi.dll')
    EnumProcesses = self.Psapi.EnumProcesses
    EnumProcesses.restype = ctypes.wintypes.BOOL

    self.EnumProcessses = EnumProcess
    ...

def getWindowsProcess(self):

    count = 50
    while True:
       .... # Build arguments to EnumProcesses and call enun process
       if self.EnumProcesses(ctypes.byref(processIds),...
       ..
       else:
           return []

我将代码分成了两个方法,以便于阅读(我相信您已经这样做了)。这里是棘手的部分,EnumProcesses正在使用指针,它们不容易处理。另一件事是,我不知道如何在Python中使用指针,所以我不能告诉您一个简单的方法来模拟它=P
我能告诉您的是,简单地不要测试它。您的逻辑非常简单。除了增加count的大小之外,该函数中的所有内容都是创建EnumProcesses指针将使用的空间。也许您可以为count大小添加限制,但除此之外,该方法简短而简洁。它仅返回窗口进程,没有其他信息。这正是我在原始评论中请求的内容:)
因此,请不要修改该方法。不要测试它。请确保使用getWindowsProcess和getLinuxProcess的任何内容都按照我的最初建议进行模拟。
希望这更有意义:)如果不是,请告诉我,也许我们可以进行聊天或视频通话。
def getWindowsProcesses(self, ...):
   '''Call Windows API function EnumProcesses and
      return the list of processes
   '''
   # ... call EnumProcesses ...
   return listOfProcesses

def getLinuxProcesses(self, ...):
   '''Look in /proc dir and return list of processes'''
   # ... look in /proc ...
   return listOfProcessses

这两种方法只做一件事情,就是获取进程列表。对于Windows系统,可能只是调用API,而对于Linux系统,只需要读取/proc目录。仅此而已,没有更多的操作。处理进程逻辑将在其他地方进行。这使得这些方法非常容易被模拟,因为它们的实现只是返回一个进程列表的API调用。

你的代码可以很容易地调用它们:

def getProcesses(...):
   '''Get the processes running.'''
   isLinux = # ... logic for determining OS ...
   if isLinux:
      processes = getLinuxProcesses(...)
   else:
      processes = getWindowsProcesses(...)
   # ... do something with processes, write to log file, etc ...

在你的测试中,你可以使用像Fudge这样的模拟库。你可以模拟这两个方法返回你期望的结果。
这样,你就可以测试你自己的逻辑,因为你可以控制结果会是什么。
from fudge import patched_context
...

def test_getProcesses(self, ...):

     monitor = MonitorTool(..)

     # Patch the method that gets the processes. Whenever it gets called, return
     # our predetermined list.
     originalProcesses = [....pids...]
     with patched_context(monitor, "getLinuxProcesses", lamba x: originalProcesses):
         monitor.getProcesses()
         # ... assert logic is right ...


     # Let's "add" some new processes and test that our logic realizes new 
     # processes were added.
     newProcesses = [...]
     updatedProcesses = originalProcessses + (newProcesses) 
     with patched_context(monitor, "getLinuxProcesses", lamba x: updatedProcesses):
         monitor.getProcesses()
         # ... assert logic caught new processes ...


     # Let's "kill" our new processes and test that our logic can handle it
     with patched_context(monitor, "getLinuxProcesses", lamba x: originalProcesses):
         monitor.getProcesses()
         # ... assert logic caught processes were 'killed' ...

请记住,如果您以这种方式测试您的代码,您将无法获得100%的代码覆盖率(因为您的模拟方法不会被执行),但这没关系。重要的是您正在测试自己的代码而不是第三方的代码。

希望这能帮助您。我知道这并没有回答您的问题,但也许您可以使用它来找出最好的测试代码的方法。


这绝对是测试我的代码的一种有趣而可行的方法。然而,我现在遇到的问题是,它并不能真正测试我的代码是否在系统上看到了新进程,而只是正确地监视了列表中的新项目。虽然代码覆盖率很重要,但我也想测试我的代码是否看到了系统上的新进程。不过还是非常感谢您的帮助! - alexquinn
@alexquinn 你是如何获取Linux进程的?你提到你正在监视/proc目录,那么你是否使用类似pyinotify的东西?另外,只是为了好玩,你能给我展示一下使用EnumProcessModules的代码行吗? - Carlos
@Carlow我会在原来的问题中发布获取系统进程的代码。 - alexquinn
我正在通过创建线程并轮询代码来监控进程,以获取系统进程(根据平台而定)。 - alexquinn
你修改后的答案对我来说非常有意义。非常感谢您花时间解释模拟这些函数背后的动机和这种测试哲学。作为一个相对新手的单元测试者,这种帮助是非常受欢迎的。 - alexquinn

0

你使用subprocess的想法很好。只需创建自己的可执行文件并将其命名为标识其为测试内容的名称。也许使它做一些暂停等操作。

或者,你可以实际使用multiprocessing模块。我在windows上很少使用python,但你应该能够从所创建的Process对象中获取进程标识数据:

p = multiprocessing.Process(target=time.sleep, args=(30,))
p.start()
pid = p.getpid()

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接