约束

pg中的约束有以下几种：

• 检查约束
• 非空约束
• 唯一约束
• 主键
• 外键

检查约束

最常见的类型，使用该约束后，在设置某个字段里的数值时必须使这个 约束的表达式的值为真。比如，限制一个人的年龄在0-150之间。

1

create table person (name varchar(40),age int check(age >=0 and age <=150),sex boolean);

也可给约束添加一个名字：

1

create table person (name varchar(40),age int constraint check_age check(age >=0 and age <=150),sex boolean);

删除约束：

1

alter table person drop constraint check_age;

重命名约束:

1

alter table person rename constraint check_age to person_age_check;

这是利用sql语句操作约束，对于我们常见的orm框架sqlalchemy应该使用CheckConstraint来添加check类型的约束。

1
2
3
4
5
6
7
8

from sqlalchemy import CheckConstraint
class Person(db.Model):
    id  = db.Column(db.Integer,primary_key=True)
    name = db.Column(db.String)
    age  = db.Column(db.Integer,CheckConstraint('age>=0 and age<=150','check_age'))
    sex = db.Column(db.Boolean)

这样我们在初始化数据结构的时候就会自动添加上对age字段的约束了。

前边两个是字段约束，我们可以在表上创建个一个约束，根据某几个字段进行约束。

语法是：

1
2
3
4
5

create table books (no integer,name varchar,price numeric,discount_price numeric, check(price>discount_price));
insert into books values (1,'test',10,12);
ERROR:  new row for relation "books" violates check constraint "books_check"
DETAIL:  Failing row contains (1, test, 10, 12).

sqlalchemy中创建表级约束的方式是:

1
2
3
4
5
6
7
8
9
10
11
12

from app import db
from sqlalchemy import CheckConstraint
class Person(db.Model):
    id  = db.Column(db.Integer,primary_key=True)
    name = db.Column(db.String)
    age  = db.Column(db.Integer)
    adult_age = db.Column(db.Integer)
    sex = db.Column(db.Boolean)
    CheckConstraint(age>adult_age)

非空约束

非空约束只是简单地声明一个字段必须不能为NULL。

1

create table books (book_no integer not null,name varchar,price numeric);

一个非空约朿总是被写成一个字段约束。非空约束在功能上等效于创建一个检査约束:

1

CHECK (column_name IS NOT NULL)

但很明显，创建一个明确的非空约束要更直观一些 。

唯一约束

唯一约束保证在一个字段或一组字段里的数据相较于表中其他行的数据是唯一的。它的语法是:

1

create table books (book_no integer unique,name varchar,price numeric);

外键约束

外键约束是表之间关系的一种约束，用于约束本表中一个字段或多个字段的数值必须出现在一个表的一个字段或多个字段中。这种约束也可以称为两个相关表之间的参照完整性约束。

如“学生表”与“班级表”之间的关系，一个学生一定是某个班级的学生：

1
2
3

CREATE TABLE class(class_no int primary key, class_name varchar(40));
CREATE TABLE student( student_no int primary key, student_name varchar(40), age int, class no int REFERENCES class(class_no);

python中的单元测试框架

python中的单元测试框架有很多，什么unittest, testtools, subunit, coverage, testrepository, nose, mox, mock, fixtures, discover等等，看着就眼花撩乱。单元测试的重要性就不多说了，作为一名合格的程序员，会写单元测试是必要条件，这里我们就从最简单的单元测试开始看起。

Unittest

Unittest是python的一个标准模块，是其他框架的基础。unittest有几个重要的概念，test fixture, test case, test suite, test runner，只有理解了这几个概念，才能真正的理解单元测试的基本原理。

• TestCase: 一个TestCase的实例就是一个测试用例。什么是测试用例呢？就是一个完整的测试流程，包括测试前准备环境的搭建(setUp)，执行测试代码(run)，以及测试后环境的还原(tearDown)。Unittest的本质也就在这里，一个测试用例是一个完整的测试单元，通过运行这个测试单元，可以对某一个问题进行验证。

• TestSuit: 而多个测试用例集合在一起，就是TestSuite，而且TestSuite也可以嵌套TestSuite

• TestLoader: 是用来加载TestCase到TestSuite中的，其中有几个loadTestsFrom__()方法，就是从各个地方寻找TestCase，创建它们的实例，然后add到TestSuite中，再返回一个TestSuite实例。

• TextTestRunner是来执行测试用例的，其中的run(test)会执行TestSuite/TestCase中的run(result)方法。

• 测试的结果会保存到TextTestResult实例中，包括运行了多少测试用例，成功了多少，失败了多少等信息

下面我们就写一个简单的测试用例。

1
2
3
4

class NumberAdd(object):
    def Add(self,num):
        return num+100

然后写一个测试用例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import unittest
from model.models import NumberAdd
class TestNum(unittest.TestCase):
    def setUp(self):
        self.num = NumberAdd()
    def testAdd(self):
        assert self.num.Add(100) == 200
    def tearDown(self):
        self.num = None
if __name__ == "__main__":
    unittest.main()

运行这个测试用例：

1
2
3
4
5
6
7
8

python3 test_num.py
kevin@qzoo:~/codes/scripts/python/untest$ python3 -m unittest discover
.
----------------------------------------------------------------------
Ran 1 test in 0.000s
OK

关于python -m

相信你一定见过python -m的用法，但是你知道python test.py 和 python -m test.py的区别吗？

python test.py: 直接运行, 就是把test.py所在的目录放到了sys.path中。

python -m test.py: 把模块当作脚本启动，也就是把你当前输入命令的目录放到sys.path中。

比如，我们有一个工程文件夹，结构如下：

1
2
3
4
5
6

model/
    __init__.py
    models.py
tests/
    __init__.py
    test_num.py

如果我们直接运行，python3 tests/test_num.py会报如下的错误：

1
2
3
4

Traceback (most recent call last):
  File "tests/test_num.py", line 3, in <module>
    from model.models import NumberAdd
ModuleNotFoundError: No module named 'model'

但是我们如果以模块的方式启动，则可以正常运行：

1
2
3
4
5
6

kevin@qzoo:~/codes/scripts/python/untest$ python3 -m tests.test_num
.
----------------------------------------------------------------------
Ran 1 test in 0.000s
OK

这就是python -m的作用。

运行unittest

可以使用命令

1

python3 -m unittest tests.test_num

也可以使用discover自动查找测试用例并执行

1
2
3
4
5
6

kevin@qzoo:~/codes/scripts/python/untest$ python3 -m unittest discover
.
----------------------------------------------------------------------
Ran 1 test in 0.001s
OK

集成Unittest到SonarQube

在CI/CD的过程中，用到了代码测试工具SonarQube，并集成到了jenkins中，这样使得在构建之前就可以使用SonarQube对代码进行单元测试。

SonarQube集成python框架可以使用nose,但是这里我们先使用最原始的unittest，看一下如何在SonarQbue中部署。

在Jenkins的构建步骤中首先配置好SonarQube:

1
2
3
4
5
6
7
8
9

sonar.projectKey=order_management_of_petrocn
sonar.projectName=order_management_of_petrocn
sonar.language=py
sonar.python.coverage.reportPath=./flask_order/coverage.xml
sonar.python.xunit.reportPath=./flask_order/nosetests.xml
sonar.projectVersion=0.1
sonar.sources=./flask_order/
sonar.sourceEncoding=UTF-8
sonar.python.coveragePlugin=cobertura

然后再shell中执行命令：

1

sudo nosetests --with-xcoverage --with-xunit --cover-package=app/tests --cover-erase --cover-tests --cover-branches -w app/tests --exe

因为nosetests命令默认是会忽略掉可执行文件的，所以要加上–exe来确保可以正常启动测试。

关于单元测试的其他框架和内容，等有时间再继续。

以前讲过如何做一个自己的python包，现在说一下如何将自己制作的包上传到pypi, 然后就可以像我们使用别的轮子一样使用自己的轮子了。

注册pypi

首先需要去 pypi.org 注册一个自己的账号和密码

生成自己的python包

使用

1
2
3
4
5
6
7

# 上传
首先需要安装twine：
```sh
sudo pip install twine

然后使用命令twine上传就可以了

1

python3 setup.py sdist upload dist/youregg.tar.gz

在上传自己的包到pypi的时候，碰到了一个错误：

1

pkg_resources.ContextualVersionConflict: (idna 2.8 (/home/kevin/.local/lib/python3.6/site-packages), Requirement.parse('idna<2.7,>=2.5') {requests}

这个错误的原因是requests库使用的idna库的版本过低。解决方法是降低idna版本或者升级requests版本即可。

有个需求需要知道自己的公网IP地址，网上的方法大概有如下几种：

1

curl ipinfo.io

通过访问第三方网站获取自己的IP，优点是简单，缺点是不够稳定，运气不好会经常访问超时。

还有一种是通过DNS

1

dig @resolver1.opendns.com ANY myip.opendns.com +short

这种方法的问题是 获取的IP未必是自己的真实的IP，不够可靠。

后来想了想，与其用别人提供的服务，不如自己写一个，原理也非常简单，就是写一个方法，通过Nginx获取到来访者的公网IP，然后将其回显到页面上。

虚拟机重启以后，突然发现做好的docker端口映射不能使用了，docker本身还在运行，但是就是访问不了服务。

通过 netstat -lntp命令查看端口使用情况:

1
2
3
4
5
6
7
8
9

(Not all processes could be identified, non-owned process info
 will not be shown, you would have to be root to see it all.)
Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address           Foreign Address         State       PID/Program name    
tcp        0      0 127.0.0.53:53           0.0.0.0:*               LISTEN      -                   
tcp        0      0 0.0.0.0:22              0.0.0.0:*               LISTEN      -                   
tcp6       0      0 :::8069                 :::*                    LISTEN      -                   
tcp6       0      0 :::8071                 :::*                    LISTEN      -                   
tcp6       0      0 :::22                   :::*                    LISTEN      -

原来端口居然变成了tcp6

禁用ipv6

ubuntu 18.04.1 禁用ipv6的方法如下：

1
2
3
4
5
6
7
8

sudo vi /etc/default/grub
GRUB_CMDLINE_LINUX_DEFAULT="ipv6.disable=1"
GRUB_CMDLINE_LINUX="ipv6.disable=1"
sudo update-grub
sudo reboot

xshell确实好用，但是不便于辗转流连于多个电脑之间，虽然用onedrive解决了多台电脑之间的秘钥文件的存储问题，但是令人讨厌的秘钥密码却总是如幽灵般的闪现，偶尔记不住某个服务器的密码就要翻箱倒柜地寻找密码薄。因此，一直有一个想法，就是搞一套webssh，像登录网站那样登录ssh，这样不论在哪台电脑上工作，只要有一个浏览器，就可以随时登录服务器，岂不快哉。

原理

首先要了解我们使用webssh的工作原理，传统单工的http协议并不能满足我们的要求，我们要求实时双工通信，因此前端的协议就是websocket，前端处理终端字符有一个非常著名的库xterm.js, 最近非常流行的vscode的terminal就是使用xterm.js编写的。后端websocket服务采用tornado, ssh通信及认证等工作交给python的paramiko库。

原理图大概如下这个样子：

xterm.js的使用

了解了原理，接下来我们来实现它。首先我们先看一下xterm.js这个工具的使用。
xterm.js是一个非常强大的库，可以帮我们简化很多的工作和麻烦，有了xterm.js，我们不需要考虑terminal下的各种字符的问题。

xterm.js的官网。

我们使用的代码比较简单，只需要引入xterm.js,然后创建一个websocket即可。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

<head>
  <link rel="stylesheet" href="{{ url_for('static', filename='dist/xterm.css') }}" />
  <link rel="stylesheet" href="{{ url_for('static', filename='dist/addons/fullscreen/fullscreen.css') }}" />
  <!-- <link rel="stylesheet" href="/dist/style.css" /> -->
  <script src="{{ url_for('static', filename='dist/xterm.js') }}"></script>
  <script src="{{ url_for('static', filename='dist/addons/fullscreen/fullscreen.js') }}"></script>
  <script src="{{ url_for('static', filename='dist/addons/fit/fit.js') }}"></script>
</head>
<body>
  <div class="container">
    <div id="terminal-container"></div>
  </div>
  <script>
    // terminado.apply(Terminal);
    console.log("加载addons");
    Terminal.applyAddon(fit);
    // 获取网页高度和宽度
    var cols = parseInt(document.documentElement.clientWidth/9.5,10)
    var rows = parseInt(document.documentElement.clientHeight/18,10)
    var term = new Terminal({
      cols: cols,
      rows: rows
    }),
      protocol = (location.protocol === 'https:') ? 'wss://' : 'ws://',
      socketURL = protocol + location.hostname + ((location.port) ? (':' + location.port) : '') + "/websocket/{{server_id}}";
    sock = new WebSocket(socketURL);
    sock.addEventListener('open', function () {
      // term(sock);
      //发送当前窗口大小
      sock.send("size:" + cols + "," + rows)
      term.on('data', function (data) {
        sock.send(data);
      });
    });
    sock.addEventListener("message", function (msg) {
      term.write(msg.data)
    });
    term.open(document.getElementById('terminal-container'));
    // term.write("Hello");
    term.fit();
    // term.toggleFullScreen(true);
  </script>
</body>

xterm.js自带了两个插件，一个fit,另一个fullscreen,使用fullscreen的时候需要注意，不要忘了引用css文件。详细使用说明请参考官网说明文档。

tornado 中使用websocket

tornado是一个强大的web应用框架，其最受关注的功能是对异步的支持，它使得处理非阻塞请求更容易，最终导致更高效的处理以及更好的可扩展性，同时也支持websocket. 所以，我们这里选用的websocket实现就是tornado。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

from tornado.web import FallbackHandler, Application, StaticFileHandler
from tornado.wsgi import WSGIContainer
from tornado.ioloop import IOLoop
from app.ws.server import SshHandler
rom tornado.httpserver import HTTPServer
import ssl
import os
app = WSGIContainer(app)
handlers = [
    (r"/websocket/(.*)", SshHandler,{}),
    (r"/(.*)", FallbackHandler, dict(fallback=app))
]
application = Application(handlers,debug=True)

这里不仅使用了websocket，还是用flask部分代码，是因为我的项目中http请求是用flask来实现的。

处理secure websocket

webssh这么敏感的业务肯定是要跑在secure websocket上的，tornado添加证书和秘钥也非常简单：

1
2
3
4
5
6
7

httpserver = HTTPServer(application,ssl_options={
            "certfile": os.path.join(config.read("secure_path"), "1_home.mixoo.cn_bundle.crt"),
            "keyfile":  os.path.join(config.read("secure_path"), "2_home.mixoo.cn.key"),
        })
httpserver.listen(int(config.read("PORT")))
    IOLoop.current().start()

这样就构建好了服务端的websocket。

paramiko

xterm.js官方使用的是termindo,一个由tornado编写而来的演示框架，但是它只支持本机的ssh，并不能满足我们希望连接其他主机的跳板需求。
这里是用的解决方案是python的paramiko库，paramiko提供很多接口供我们使用，能让我们轻松地完成ssh连接操作。

通常我们远程登录主机使用的验证方式有两种，一种是用户名和密码认证，另一种是私钥公钥认证，有时候私钥还会有密码。这些在xshell等工具中都有配置，我们现在就用paramiko来实现同样的功能。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

class SSH(object):
    def __init__(self, host, port, user, password=None, keyfile=None, passphrase=None):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.keyfile = keyfile
        self._ssh = paramiko.SSHClient()
        self._ssh.load_system_host_keys()
        self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        k = keyfile and paramiko.RSAKey.from_private_key_file(
            keyfile, password=passphrase) or None
        self._ssh.connect(hostname=host, port=port, username=user,
                          password=password, pkey=k)
        self._chanel = self._ssh.invoke_shell(
            term='xterm')
    def resize(self, cols, rows):
        self._chanel.resize_pty(width=cols, height=rows)
    def send(self, msg):
        self._chanel.send(msg)
    def read(self):
        return self._chanel.recv(10000)

核心代码不多，主要的方法就是connect，在其中要将用户名和密码、端口等信息都配置好。如果有私钥，私钥又有密码，就使用from_private_key_file方法把私钥文件名和密码传入。invoke_shell方法返回一个channel，拿到这个channal就可以像xshell一样打开一个terminal一样来操作了。

跳坑指南

Nignx部署400错误

本地测试完，部署服务器的时候碰到nginx返回400错误。原因是忘了在nginx配置中将协议升级为websocket:

1
2
3

proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection $connection_upgrade;

RuntimeError: There is no current event loop in thread ‘Thread-1’.

也是在部署服务器的时候出现的，原因是本地tornado版本是4.5.3，而服务器版本变成了5.1. 在5.1版本中使用多线程，就会出现上述错误。解决方案就是使用asyncio的set_event_loop方法：

1
2
3
4
5
6

import asyncio
def _reading(self):
        asyncio.set_event_loop(asyncio.new_event_loop())
        while True:
            data = self.ssh.read()
            self.write_message(data)

Websocket报403错误

也是因为服务器版本5.1的问题，解决方案，在websocket的handler中添加一个check_origin方法：

1
2

def check_origin(self, origin):
    return True

运行界面：

点击终端进入webssh界面：

源代码地址

要继续使用此程序，您必须应用最新的更新或使用新版本

用的好好的xshell 5，突然间就不让用了，提示什么必须要在更新再使用。鉴于之前用过xshell6，对于免费版只能同时开4个tab的限制实在不能忍受，才退回了xshell 5. 看来官方这是要把人逼上梁山啊。

破解方法

下载 ns.dll 放到xshell的安装目录，覆盖即可。

在使用树莓派上传视频到七牛云的时候，在本地电脑上可以正常上传，但是放到树莓派上就发现DNS解析错误

使用cat /etc/resolv.conf查看树莓派的DNS设置：

1
2
3

nameserver 114.114.114.114
nameserver 220.x.x.x

看起来没有什么问题，可就是ping不同七牛的域名uc.qbox.me

后来发现uc.qbox.me的服务器在北美，可能国内某些dns解析不成功。
于是给树莓派换DNS服务器就成了必须要的一步：

编辑 /etc/resolvconf.conf：

加入一条：

1

name_servers=8.8.8.8

然后重启树莓派即可。

自定义安装包

在之前的文章中，我们介绍过如何自定义一个安装包，并制作自己的启动脚本，但是之前的步骤中存在一个问题，就是如果我们自定义了Install脚本，将无法自动安装依赖。

解决方案

自动安装依赖的脚本实际是在Install类中调用了easy_install,我们可以在我们的脚本中调用该方法实现手动安装依赖的效果。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

setup(name='MoYao-Raspberry-Controller',
      version=moyao.__version__,
      description='Python Distribution Utilities',
      author='Kevin Kong',
      author_email='kfxw2007@163.com',
      license='http://www.apache.org/licenses/LICENSE-2.0.html',
      url='',
      data_files=data_files,
      packages=find_packages(),
      package_dir={'%s' % 'moyao': 'moyao'},
      include_package_data=True,
      python_requires='>=3.5',
      install_requires=[
          "pywifi",
          "autils",
          "picamera",
          "configparser",
          "pyserial",
          "pexpect",
          "requests",
          "pywifi",
          "paho-mqtt",
          "qiniu"
      ],
      cmdclass={
          'install': InstallCommand,
      },
    )

在Install脚本中执行安装依赖：

1
2
3
4

def run(self):
        install.run(self)
        install.do_egg_install(self)
        ...