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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
#! /usr/bin/env python
# -*- tab-width: 4; indent-tabs-mode: nil; py-indent-offset: 4 -*-
#
# This file is part of the LibreOffice project.
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
#
import unittest
import uno
from org.libreoffice.unotest import UnoInProcess
class TestTempFile(unittest.TestCase):
"""Test temporary file object created from com.sun.star.io.TempFile"""
@classmethod
def setUpClass(cls):
cls._uno = UnoInProcess()
cls._uno.setUp()
@classmethod
def tearDownClass(cls):
cls._uno.tearDown()
def setUp(self):
self.file_data = uno.ByteSequence(b"some data")
service_manager = self._uno.getContext().getServiceManager()
if service_manager is None:
raise RuntimeError("Cannot create service factory!")
try:
self.file_access = service_manager.createInstance("com.sun.star.ucb.SimpleFileAccess")
if self.file_access is None:
raise RuntimeError("Cannot get simple access!")
except Exception as e:
raise RuntimeError(f"Cannot get simple file access! {e}")
self.temp_file = service_manager.createInstance("com.sun.star.io.TempFile")
has_xtempfile_if = bool([
1 for type_info in self.temp_file.getTypes()
if type_info.typeName == "com.sun.star.io.XTempFile"
])
if not has_xtempfile_if:
raise RuntimeError("Cannot get XTempFile interface.")
def close_temp_file(self) -> None:
stream = self.temp_file.getOutputStream()
if stream is None:
raise RuntimeError("Cannot get output stream")
stream.closeOutput()
stream = self.temp_file.getInputStream()
if stream is None:
raise RuntimeError("Cannot get input stream")
stream.closeInput()
print("Tempfile closed successfully.")
def read_bytes_with_stream(self) -> uno.ByteSequence:
input_stream = self.temp_file.getInputStream()
if input_stream is None:
raise RuntimeError("Cannot get input stream from tempfile.")
nbytes, read_data = input_stream.readBytes(None, len(self.file_data))
print("Read", nbytes, "bytes from tempfile successfully.")
return read_data
def read_directly_from_temp_file(self, file_url: str) -> uno.ByteSequence:
print("Attempting to read directly from", file_url)
input_stream = self.file_access.openFileRead(file_url)
if input_stream is None:
raise RuntimeError("Cannot create input stream from URL.")
nbytes, read_data = input_stream.readBytes(None, len(self.file_data))
print("Read", nbytes, "bytes directly from tempfile successfully.")
return read_data
def write_bytes_with_stream(self) -> None:
output_stream = self.temp_file.getOutputStream()
if output_stream is None:
raise RuntimeError("Cannot get output stream.")
output_stream.writeBytes(self.file_data)
output_stream.flush()
print("Write", len(self.file_data), "bytes to tempfile successfully.")
def get_temp_file_url(self) -> str:
uri = self.temp_file.Uri
if not uri:
raise RuntimeError("Temporary file not valid.")
return uri
def get_temp_file_name(self) -> str:
file_name = self.temp_file.ResourceName
if not file_name:
raise RuntimeError("Temporary file not valid.")
return file_name
def test_01(self):
file_uri = self.get_temp_file_url()
file_name = self.get_temp_file_name()
print("Tempfile URL:", file_uri)
print("Tempfile name:", file_name)
self.assertTrue(
file_uri.endswith(file_name.replace("\\", "/")),
"FILE NAME AND URL DO NOT MATCH.",
)
# write to the stream using the service.
self.write_bytes_with_stream()
# check the result by reading from the service.
self.temp_file.seek(0)
read_data = self.read_bytes_with_stream()
self.assertEqual(self.file_data, read_data, "Tempfile outputs false data!")
# check the result by reading from the file directly.
read_data = self.read_directly_from_temp_file(file_uri)
self.assertEqual(self.file_data, read_data, "Tempfile contains false data!")
# close the object(by closing input and output), check that the file
# was removed.
self.temp_file.RemoveFile = False
# After tempfile is closed, file name cannot be got from a TempFile object.
file_name = self.temp_file.ResourceName
self.close_temp_file()
self.assertTrue(
self.file_access.exists(file_name), "TempFile mistakenly removed.",
)
# Finally, cleanup this temp file.
self.file_access.kill(file_name)
def test_02(self):
self.write_bytes_with_stream()
file_url = self.get_temp_file_url()
# let the service not to remove the URL.
self.temp_file.RemoveFile = False
# close the tempfile by closing input and output.
self.close_temp_file()
# check that the file is still available.
read_data = self.read_directly_from_temp_file(file_url)
self.assertEqual(self.file_data, read_data, "Tempfile contains false data!")
if __name__ == '__main__':
unittest.main()
# vim: set shiftwidth=4 softtabstop=4 expandtab:
|
