- r - 以节省内存的方式增长 data.frame
- ruby-on-rails - ruby/ruby on rails 内存泄漏检测
- android - 无法解析导入android.support.v7.app
- UNIX 域套接字与共享内存(映射文件)
我正在为一些依赖于写入 SQLite3 数据库的高阶逻辑编写单元测试。为此,我使用了 twisted.trial.unittest
和 twisted.enterprise.adbapi.ConnectionPool
。
我能够创建一个持久的 sqlite3 数据库并在其中存储数据。使用 sqlitebrowser ,我能够验证数据是否已按预期持久化。
问题是调用 t.e.a.ConnectionPool.run*
(例如:runQuery
)返回一组空结果,但仅当从 中调用时测试用例
.
我遇到的问题只发生在 Twisted 的 trial
框架内。我的第一次调试尝试是将数据库代码从单元测试中拉出来,并将其放入一个独立的测试/调试脚本中。所述脚本按预期工作,而单元测试代码没有(参见下面的示例)。
这是用来初始化数据库的脚本。此文件没有(明显的)错误。
CREATE TABLE ajxp_changes ( seq INTEGER PRIMARY KEY AUTOINCREMENT, node_id NUMERIC, type TEXT, source TEXT, target TEXT, deleted_md5 TEXT );
CREATE TABLE ajxp_index ( node_id INTEGER PRIMARY KEY AUTOINCREMENT, node_path TEXT, bytesize NUMERIC, md5 TEXT, mtime NUMERIC, stat_result BLOB);
CREATE TABLE ajxp_last_buffer ( id INTEGER PRIMARY KEY AUTOINCREMENT, type TEXT, location TEXT, source TEXT, target TEXT );
CREATE TABLE ajxp_node_status ("node_id" INTEGER PRIMARY KEY NOT NULL , "status" TEXT NOT NULL DEFAULT 'NEW', "detail" TEXT);
CREATE TABLE events (id INTEGER PRIMARY KEY AUTOINCREMENT, type text, message text, source text, target text, action text, status text, date text);
CREATE TRIGGER LOG_DELETE AFTER DELETE ON ajxp_index BEGIN INSERT INTO ajxp_changes (node_id,source,target,type,deleted_md5) VALUES (old.node_id, old.node_path, "NULL", "delete", old.md5); END;
CREATE TRIGGER LOG_INSERT AFTER INSERT ON ajxp_index BEGIN INSERT INTO ajxp_changes (node_id,source,target,type) VALUES (new.node_id, "NULL", new.node_path, "create"); END;
CREATE TRIGGER LOG_UPDATE_CONTENT AFTER UPDATE ON "ajxp_index" FOR EACH ROW BEGIN INSERT INTO "ajxp_changes" (node_id,source,target,type) VALUES (new.node_id, old.node_path, new.node_path, CASE WHEN old.node_path = new.node_path THEN "content" ELSE "path" END);END;
CREATE TRIGGER STATUS_DELETE AFTER DELETE ON "ajxp_index" BEGIN DELETE FROM ajxp_node_status WHERE node_id=old.node_id; END;
CREATE TRIGGER STATUS_INSERT AFTER INSERT ON "ajxp_index" BEGIN INSERT INTO ajxp_node_status (node_id) VALUES (new.node_id); END;
CREATE INDEX changes_node_id ON ajxp_changes( node_id );
CREATE INDEX changes_type ON ajxp_changes( type );
CREATE INDEX changes_node_source ON ajxp_changes( source );
CREATE INDEX index_node_id ON ajxp_index( node_id );
CREATE INDEX index_node_path ON ajxp_index( node_path );
CREATE INDEX index_bytesize ON ajxp_index( bytesize );
CREATE INDEX index_md5 ON ajxp_index( md5 );
CREATE INDEX node_status_status ON ajxp_node_status( status );
这是意外失败的单元测试类。 TestStateManagement.test_db_clean
通过,表明表已正确创建。 TestStateManagement.test_inode_create
失败,报告检索到零个结果。
import os.path as osp
from twisted.internet import defer
from twisted.enterprise import adbapi
import sqlengine # see below
class TestStateManagement(TestCase):
def setUp(self):
self.meta = mkdtemp()
self.db = adbapi.ConnectionPool(
"sqlite3", osp.join(self.meta, "db.sqlite"), check_same_thread=False,
)
self.stateman = sqlengine.StateManager(self.db)
with open("init.sql") as f:
script = f.read()
self.d = self.db.runInteraction(lambda c, s: c.executescript(s), script)
def tearDown(self):
self.db.close()
del self.db
del self.stateman
del self.d
rmtree(self.meta)
@defer.inlineCallbacks
def test_db_clean(self):
"""Canary test to ensure that the db is initialized in a blank state"""
yield self.d # wait for db to be initialized
q = "SELECT name FROM sqlite_master WHERE type='table' AND name=?;"
for table in ("ajxp_index", "ajxp_changes"):
res = yield self.db.runQuery(q, (table,))
self.assertTrue(
len(res) == 1,
"table {0} does not exist".format(table)
)
@defer.inlineCallbacks
def test_inode_create_file(self):
yield self.d
path = osp.join(self.ws, "test.txt")
with open(path, "wt") as f:
pass
inode = mk_dummy_inode(path)
yield self.stateman.create(inode, directory=False)
entry = yield self.db.runQuery("SELECT * FROM ajxp_index")
emsg = "got {0} results, expected 1. Are canary tests failing?"
lentry = len(entry)
self.assertTrue(lentry == 1, emsg.format(lentry))
这些是上述单元测试正在测试的工件。
def values_as_tuple(d, *param):
"""Return the values for each key in `param` as a tuple"""
return tuple(map(d.get, param))
class StateManager:
"""Manages the SQLite database's state, ensuring that it reflects the state
of the filesystem.
"""
log = Logger()
def __init__(self, db):
self._db = db
def create(self, inode, directory=False):
params = values_as_tuple(
inode, "node_path", "bytesize", "md5", "mtime", "stat_result"
)
directive = (
"INSERT INTO ajxp_index (node_path,bytesize,md5,mtime,stat_result) "
"VALUES (?,?,?,?,?);"
)
return self._db.runOperation(directive, params)
twisted.trial
之外消失了#! /usr/bin/env python
import os.path as osp
from tempfile import mkdtemp
from twisted.enterprise import adbapi
from twisted.internet.task import react
from twisted.internet.defer import inlineCallbacks
INIT_FILE = "example.sql"
def values_as_tuple(d, *param):
"""Return the values for each key in `param` as a tuple"""
return tuple(map(d.get, param))
def create(db, inode):
params = values_as_tuple(
inode, "node_path", "bytesize", "md5", "mtime", "stat_result"
)
directive = (
"INSERT INTO ajxp_index (node_path,bytesize,md5,mtime,stat_result) "
"VALUES (?,?,?,?,?);"
)
return db.runOperation(directive, params)
def init_database(db):
with open(INIT_FILE) as f:
script = f.read()
return db.runInteraction(lambda c, s: c.executescript(s), script)
@react
@inlineCallbacks
def main(reactor):
meta = mkdtemp()
db = adbapi.ConnectionPool(
"sqlite3", osp.join(meta, "db.sqlite"), check_same_thread=False,
)
yield init_database(db)
# Let's make sure the tables were created as expected and that we're
# starting from a blank slate
res = yield db.runQuery("SELECT * FROM ajxp_index LIMIT 1")
assert not res, "database is not empty [ajxp_index]"
res = yield db.runQuery("SELECT * FROM ajxp_changes LIMIT 1")
assert not res, "database is not empty [ajxp_changes]"
# The details of this are not important. Suffice to say they (should)
# conform to the DB schema for ajxp_index.
test_data = {
"node_path": "/this/is/some/arbitrary/path.ext",
"bytesize": 0,
"mtime": 179273.0,
"stat_result": b"this simulates a blob of raw binary data",
"md5": "d41d8cd98f00b204e9800998ecf8427e", # arbitrary
}
# store the test data in the ajxp_index table
yield create(db, test_data)
# test if the entry exists in the db
entry = yield db.runQuery("SELECT * FROM ajxp_index")
assert len(entry) == 1, "got {0} results, expected 1".format(len(entry))
print("OK")
同样,在使用 sqlitebrowser 检查时,数据似乎正在写入 db.sqlite
,因此这看起来像是一个检索 问题。从这里开始,我有点难过……有什么想法吗?
此代码将生成可用于测试的inode
。
def mk_dummy_inode(path, isdir=False):
return {
"node_path": path,
"bytesize": osp.getsize(path),
"mtime": osp.getmtime(path),
"stat_result": dumps(stat(path), protocol=4),
"md5": "directory" if isdir else "d41d8cd98f00b204e9800998ecf8427e",
}
最佳答案
好吧,事实证明这是一个有点棘手的问题。单独运行测试(如发布到此问题)使得错误很少发生。但是,当在整个测试套件的上下文中运行时,它几乎 100% 都会失败。
我在写入数据库之后和从数据库读取之前添加了 yield task.deferLater(reactor, .00001, lambda: None)
,这解决了问题。
从那里,我怀疑这可能是连接池和 sqlite 有限的并发容忍度引起的竞争条件。我尝试将 cb_min
和 cb_max
参数设置为 ConnectionPool
为 1
,这也解决了这个问题。
简而言之:似乎 sqlite 不能很好地处理多个连接,适当的解决方法是尽可能避免并发。
关于python - 为什么 Twisted 的 adbapi 无法从单元测试中恢复数据?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/44160375/
我正在尝试发送 RFC5424 中定义的结构化数据从 Twisted 到系统日志。我在 Ubuntu 上,系统日志实现是 rsyslog。 我正在使用“twistd --syslog”运行扭曲的应用程
摘要:无法自动向 Twisted 的检修孔发出命令。寻求解决方案/建议。 Twisted 有一个很棒的功能,称为 manhole。它允许用户通过 ssh 连接到当前正在运行的 Twisted 服务器并
我正在尝试发送 RFC5424 中定义的结构化数据从 Twisted 到系统日志。我在 Ubuntu 上,系统日志实现是 rsyslog。 我正在使用“twistd --syslog”运行扭曲的应用程
我正在尝试使用扭曲的 spawnProcess 启动一个进程,并使用 psutil 每 5 秒记录一次资源使用情况。首先,我尝试使用以下代码获取生成的进程的可执行文件名称: #!/usr/bin/py
Azure 网站能否托管 Twisted 应用程序?例如像这样的东西: from twisted.internet import reactor from twisted.web import ser
有没有办法限制 Twisted http 客户端的下载速率?如果没有,在 Twisted 中实现此类客户端的最简单方法是什么? 最佳答案 Twisted 中的流控制最常使用 IProducer.pau
我的基于 Twisted 的客户端循环发送 UDP 数据包。因此,我正在使用 DatagramProtocol 类。这是来源: #!/usr/bin/python # -*- coding: utf-
raven 有很多集成,包括 python 日志记录。一方面,twisted 不使用python 的日志记录。而另一方面,在twisted 中,raven 并没有直接的整合。 那么在基于扭曲的设置中使
有人告诉我,Twisted基于库Ampoule是创建在不同计算机上执行的进程池的好方法。然而,没有相关文档,Ampoule 的示例也没有说明这一点。 我对类似于 stdlib multiprocess
Twisted 最近推出了一个新的日志记录模块:twisted.logger。我已经通读了文档 [1],但我一直无法找到设置实际日志级别的位置。关于日志观察者[2] 的部分建议,如果您显式配置自己的
我想问一个关于如何关闭扭曲连接的问题 RPC . 我知道有人问过类似的问题,但似乎没有回答我的问题。 我正在做一些基本的连接,如下图所示: cfactory = pb.PBClientFactory(
我正在使用 Twisted 编写代码,但在为我的扭曲互联网延迟变量想出一个合理的变量名时遇到了麻烦。这是我的候选人: d :太通用,太短,违反了 pylint 规则 C0103。 def :与内置函数
我正在为 Web 服务器的静态部分使用 Twisted Web static.File 资源。 对于开发,我希望能够添加新文件或修改当前静态文件,而无需重新启动 Twisted 网络服务器。 我在 g
我想使用 Twisted 重建现有应用程序的通信部分。这个应用程序确实从客户端向服务器发送数据,只是这样,服务器不发送任何东西。 我如何使用 Twisted 的事件驱动概念来实现这一目标?我目前使用
我开始考虑通过扩展当前的 Twisted FTP 来实现显式 FTP。 大部分代码都很简单,实现 AUTH、PBSZ、PROT 很容易,我得到了一个有效的安全控制 channel 。 我的问题是数据通
我想实现 WebRTC 对等连接。我不知道如何设置 ICE 服务器或应该使用什么工具。任何人都可以给我建议吗?非常感谢。 最佳答案 实际上你不需要编写自己的 STUN/TURN 服务器 有很多现成的解
正如标题所示,ProcessProtocol 类上的这两个函数有什么区别?关于什么时候应该使用一种而不是另一种的文档有点稀疏? 我最好寻找一些可以证明这一点的用例示例。 最佳答案 我猜文档在这一点上有
使用下面的代码,我似乎可以相当轻松地使用 multiprocessing.reduction 在子进程中重建套接字.. import socket,os import multiprocessing
我正在尝试在 python 上使用 twisted 编写服务器。这是我文件的头部: from twisted.internet.protocol import Factory, Protocol fr
我使用 Twisted 制作了一个简单的 http 服务器,它发送 Content-Type: multipart/x-mixed-replace header 。我正在使用它来测试我想设置为接受长期
我是一名优秀的程序员,十分优秀!