3415
+ − 1 #!/usr/bin/env python3
+ − 2
+ − 3 # SàT: an XMPP client
+ − 4 # Copyright (C) 2009-2020 Jérôme Poisson (goffi@goffi.org)
+ − 5
+ − 6 # This program is free software: you can redistribute it and/or modify
+ − 7 # it under the terms of the GNU Affero General Public License as published by
+ − 8 # the Free Software Foundation, either version 3 of the License, or
+ − 9 # (at your option) any later version.
+ − 10
+ − 11 # This program is distributed in the hope that it will be useful,
+ − 12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
+ − 13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ − 14 # GNU Affero General Public License for more details.
+ − 15
+ − 16 # You should have received a copy of the GNU Affero General Public License
+ − 17 # along with this program. If not, see <http://www.gnu.org/licenses/>.
+ − 18
+ − 19 import os
+ − 20 import shutil
+ − 21 import pytest
+ − 22 import sh
+ − 23 from sh import jp
+ − 24 from sat.tools.common import uri
+ − 25
+ − 26
+ − 27 if os . getenv ( "SAT_TEST_ENV_E2E" ) is None :
+ − 28 pytest . skip (
+ − 29 "skipping end-to-end tests, we are not in a test environment" ,
+ − 30 allow_module_level = True
+ − 31 )
+ − 32
+ − 33
+ − 34 pytestmark = pytest . mark . usefixtures ( "test_profiles" )
+ − 35
+ − 36
+ − 37 class TestInstall :
+ − 38
+ − 39 def test_jp_can_run ( self ):
+ − 40 jp ( "--version" )
+ − 41
+ − 42
+ − 43 class TestJpAccount :
+ − 44
+ − 45 def test_create_and_delete ( self , jp_json ):
+ − 46 """Create an account in-band, connect it, then delete it and its profile"""
+ − 47 jp . account . create (
+ − 48 "test_create@server1.test" ,
+ − 49 "test" ,
+ − 50 profile = "test_create" ,
+ − 51 host = "server1.test"
+ − 52 )
+ − 53 profiles = jp_json . profile . list ()
+ − 54 assert "test_create" in profiles
+ − 55 jp . profile . connect ( connect = True , profile = "test_create" )
+ − 56 jp . account . delete ( profile = "test_create" , force = True )
+ − 57 jp . profile . delete ( "test_create" , force = True )
+ − 58 profiles = jp_json . profile . list ()
+ − 59 assert "test_create" not in profiles
+ − 60
+ − 61
+ − 62 @pytest . mark . usefixtures ( "pubsub_nodes" )
+ − 63 class TestJpPubsub :
+ − 64
+ − 65 def test_node_create_info_delete ( self ):
+ − 66 node_name = "tmp_node"
+ − 67 with pytest . raises ( sh . ErrorReturnCode_16 ):
+ − 68 # the node should not exist
+ − 69 jp . pubsub . node . info ( node = node_name )
+ − 70 try :
+ − 71 jp . pubsub . node . create ( node = node_name )
+ − 72 # if node exist has expected, following command won't raise an exception
+ − 73 metadata = jp . pubsub . node . info ( node = node_name )
+ − 74 assert len ( metadata . strip ())
+ − 75 finally :
+ − 76 jp . pubsub . node . delete ( node = node_name , force = True )
+ − 77
+ − 78 with pytest . raises ( sh . ErrorReturnCode_16 ):
+ − 79 # the node should not exist anymore
+ − 80 jp . pubsub . node . info ( node = node_name )
+ − 81
+ − 82 def test_set_get_delete_purge ( self , jp_elt ):
+ − 83 content = "test item"
+ − 84 payload = f "<test> { content } </test>"
+ − 85
+ − 86 # we create 3 items and check them
+ − 87 item1_id = jp . pubsub . set ( node = "test" , quiet = True , _in = payload )
+ − 88 item2_id = jp . pubsub . set ( node = "test" , quiet = True , _in = payload )
+ − 89 item3_id = jp . pubsub . set ( node = "test" , quiet = True , _in = payload )
+ − 90 parsed_elt = jp_elt . pubsub . get ( node = "test" , item = item1_id )
+ − 91 payload = parsed_elt . firstChildElement ()
+ − 92 assert payload . name == 'test'
+ − 93 assert str ( payload ) == content
+ − 94 parsed_elt = jp_elt . pubsub . get ( node = "test" , item = item2_id )
+ − 95 payload = parsed_elt . firstChildElement ()
+ − 96 assert payload . name == 'test'
+ − 97 assert str ( payload ) == content
+ − 98 parsed_elt = jp_elt . pubsub . get ( node = "test" , item = item3_id )
+ − 99 payload = parsed_elt . firstChildElement ()
+ − 100 assert payload . name == 'test'
+ − 101 assert str ( payload ) == content
+ − 102
+ − 103 # deleting first item should work
+ − 104 jp . pubsub . delete ( node = "test" , item = item1_id , force = True )
+ − 105 with pytest . raises ( sh . ErrorReturnCode_16 ):
+ − 106 jp . pubsub . get ( node = "test" , item = item1_id )
+ − 107
+ − 108 # there must be a least item2 and item3 in the node
+ − 109 node_items = jp_elt . pubsub . get ( node = "test" )
+ − 110 assert len ( list ( node_items . elements ())) >= 2
+ − 111
+ − 112 # after purge, node must be empty
+ − 113 jp . pubsub . node . purge ( node = "test" , force = True )
+ − 114 node_items = jp_elt . pubsub . get ( node = "test" )
+ − 115 assert len ( list ( node_items . elements ())) == 0
+ − 116
+ − 117 def test_edit ( self , editor , jp_elt ):
+ − 118 content = "original item"
+ − 119 payload = f "<test> { content } </test>"
+ − 120 item_id = jp . pubsub . set ( node = "test" , quiet = True , _in = payload )
+ − 121 editor . set_filter ( 'content.replace("original", "edited")' )
+ − 122 jp . pubsub . edit ( node = "test" , item = item_id , _env = editor . env )
+ − 123 assert "original item" in editor . original_content
+ − 124 parsed_elt = jp_elt . pubsub . get ( node = "test" , item = item_id )
+ − 125 edited_payload = parsed_elt . firstChildElement ()
+ − 126 expected_edited_content = content . replace ( "original" , "edited" )
+ − 127 assert edited_payload . name == 'test'
+ − 128 assert str ( edited_payload ) == expected_edited_content
+ − 129
+ − 130 def test_affiliations ( self , jp_json ):
+ − 131 affiliations = jp_json . pubsub . affiliations ()
+ − 132 assert affiliations [ "test" ] == "owner"
+ − 133
+ − 134 def test_uri ( self ):
+ − 135 built_uri = jp . pubsub . uri (
+ − 136 service = "pubsub.example.net" , node = "some_node"
+ − 137 ) . strip ()
+ − 138 assert built_uri == "xmpp:pubsub.example.net?;node=some_node"
+ − 139 built_uri = jp . pubsub . uri (
+ − 140 service = "pubsub.example.net" , node = "some_node" , item = "some_item"
+ − 141 ) . strip ()
+ − 142 assert built_uri == "xmpp:pubsub.example.net?;node=some_node;item=some_item"
+ − 143
+ − 144
+ − 145 class TestJpBlog :
+ − 146 MICROBLOG_NS = "urn:xmpp:microblog:0"
+ − 147
+ − 148 def test_set_get ( self , jp_json ):
+ − 149 jp . blog . set ( _in = "markdown **bold** [link](https://example.net)" )
+ − 150 item_data = jp_json . blog . get ( max = 1 )
+ − 151 item = item_data [ 0 ][ 0 ]
+ − 152 metadata = item_data [ 1 ]
+ − 153 assert metadata [ 'service' ] == "account1@server1.test"
+ − 154 assert metadata [ 'node' ] == self . MICROBLOG_NS
+ − 155 assert metadata [ 'rsm' ] . keys () <= { "first" , "last" , "index" , "count" }
+ − 156 item_id = item [ 'id' ]
+ − 157 expected_uri = uri . buildXMPPUri (
+ − 158 'pubsub' , subtype = "microblog" , path = "account1@server1.test" ,
+ − 159 node = self . MICROBLOG_NS , item = item_id
+ − 160 )
+ − 161 assert item [ 'uri' ] == expected_uri
+ − 162 assert item [ 'content_xhtml' ] == (
+ − 163 '<div><p>markdown <strong>bold</strong> '
+ − 164 '<a href="https://example.net">link</a></p></div>'
+ − 165 )
+ − 166 assert isinstance ( item [ 'published' ], int )
+ − 167 assert isinstance ( item [ 'updated' ], int )
+ − 168 assert isinstance ( item [ 'comments' ], list )
+ − 169 assert isinstance ( item [ 'tags' ], list )
+ − 170 assert item [ 'author' ] == 'account1'
+ − 171 assert item [ 'author_jid' ] == 'account1@server1.test'
+ − 172
+ − 173 def test_edit ( self , editor , jp_json ):
+ − 174 payload_md = "content in **markdown**"
+ − 175 editor . set_filter ( repr ( payload_md ))
+ − 176 jp . blog . edit ( _env = editor . env )
+ − 177 assert len ( editor . original_content ) == 0
+ − 178 assert editor . new_content == payload_md
+ − 179 items_data = jp_json . blog . get ( max = 1 )
+ − 180 last_item = items_data [ 0 ][ 0 ]
+ − 181 last_item_id = last_item [ 'id' ]
+ − 182 assert last_item [ 'content' ] == "content in markdown"
+ − 183 assert last_item [ 'content_xhtml' ] == (
+ − 184 "<div><p>content in <strong>markdown</strong></p></div>"
+ − 185 )
+ − 186 editor . set_filter ( 'f" {content} extended"' )
+ − 187 jp . blog . edit ( "--last-item" , _env = editor . env )
+ − 188 assert editor . original_content == payload_md
+ − 189 assert editor . new_content == f " { payload_md } extended"
+ − 190 items_data = jp_json . blog . get ( max = 1 )
+ − 191 last_item = items_data [ 0 ][ 0 ]
+ − 192 # we check that the id hasn't been modified
+ − 193 assert last_item [ 'id' ] == last_item_id
+ − 194 assert last_item [ 'content' ] == "content in markdown extended"
+ − 195 assert last_item [ 'content_xhtml' ] == (
+ − 196 "<div><p>content in <strong>markdown</strong> extended</p></div>"
+ − 197 )
+ − 198
+ − 199
+ − 200 class TestJpFile :
+ − 201
+ − 202 def test_upload_get ( self , fake_file ):
+ − 203 source_file = fake_file . size ( 10240 )
+ − 204 source_file_hash = fake_file . get_source_hash ( source_file )
+ − 205 upload_url = jp . file . upload ( source_file ) . strip ()
+ − 206
+ − 207 dest_file = fake_file . new_dest_file ()
+ − 208 try :
+ − 209 jp . file . get ( upload_url , dest_file = dest_file )
+ − 210 dest_file_hash = fake_file . get_dest_hash ( dest_file )
+ − 211 finally :
+ − 212 dest_file . unlink ()
+ − 213
+ − 214 assert source_file_hash == dest_file_hash
+ − 215
+ − 216 def test_send_receive ( self , fake_file ):
+ − 217 source_file = fake_file . size ( 10240 )
+ − 218 source_file_hash = fake_file . get_source_hash ( source_file )
+ − 219 send_cmd = jp . file . send ( source_file , "account1@server2.test" , _bg = True )
+ − 220 dest_path = fake_file . dest_files / "test_send_receive"
+ − 221 dest_path . mkdir ()
+ − 222 try :
+ − 223 jp . file . receive (
+ − 224 "account1@server1.test" , profile = "account1_s2" , path = dest_path )
+ − 225 dest_file = dest_path / source_file . name
+ − 226 dest_file_hash = fake_file . get_dest_hash ( dest_file )
+ − 227 finally :
+ − 228 shutil . rmtree ( dest_path )
+ − 229 send_cmd . wait ()
+ − 230
+ − 231 assert source_file_hash == dest_file_hash
