Skip to content

Commit

Permalink
fix: regression in module due to namespace changes (#2708)
Browse files Browse the repository at this point in the history
  • Loading branch information
marcelotduarte authored Nov 29, 2024
1 parent 90669bf commit 443ca14
Show file tree
Hide file tree
Showing 5 changed files with 197 additions and 25 deletions.
47 changes: 23 additions & 24 deletions cx_Freeze/module.py
Original file line number Diff line number Diff line change
Expand Up @@ -254,38 +254,37 @@ def stub_code(self) -> CodeType | None:
if ext not in EXTENSION_SUFFIXES or self.root_dir is None:
return None

package = filename.parent.relative_to(self.root_dir)
try:
package = filename.parent.relative_to(self.root_dir)
except ValueError:
return None
stem = filename.name.removesuffix(ext)
stub_name = f"{stem}.pyi"
# search for the stub file already parsed in the distribution
importshed = Path(__file__).resolve().parent / "importshed"
source_file = importshed / package / stub_name
imports_only = None
if source_file.exists():
imports_only = source_file.read_text(encoding="utf_8")
if imports_only:
return compile(
imports_only, stub_name, "exec", dont_inherit=True
)
# search for a stub file along side the python extension module
source_file = filename.parent / stub_name
if not source_file.exists():
return None
if cache_path:
target_file = cache_path / package / stub_name
if target_file.exists():
# a parsed stub exists in the cache
imports_only = target_file.read_text(encoding="utf_8")
else:
if not imports_only:
# search for a stub file along side the python extension module
source_file = filename.parent / stub_name
if source_file.exists():
imports_only = self.get_imports_from_file(source_file)
if imports_only:
# cache the parsed stub
target_file.parent.mkdir(parents=True, exist_ok=True)
target_file.write_text(
"# Generated by cx_Freeze\n\n" + imports_only,
encoding="utf_8",
)
else:
imports_only = self.get_imports_from_file(source_file)
if not imports_only and cache_path:
target_file = cache_path / package / stub_name
if target_file.exists():
# a parsed stub exists in the cache
imports_only = target_file.read_text(encoding="utf_8")
else:
imports_only = self.get_imports_from_file(source_file)
if imports_only:
# cache the parsed stub
target_file.parent.mkdir(parents=True, exist_ok=True)
target_file.write_text(
"# Generated by cx_Freeze\n\n" + imports_only,
encoding="utf_8",
)
if imports_only:
return compile(imports_only, stub_name, "exec", dont_inherit=True)
return None
Expand Down
12 changes: 12 additions & 0 deletions samples/win32com/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,12 @@
# Samples

Here are samples to test cx_Freeze or to show how to use a package in cx_Freeze.

# Installation and requirements:

In a virtual environment, install by issuing the command:

```
pip install --upgrade cx_Freeze pywin32
cxfreeze --script test_win32com.py --silent
```
107 changes: 107 additions & 0 deletions samples/win32com/test_win32com.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,107 @@
"""Implements _both_ a connectable client, and a connectable server.
https://brian3johnson.github.io/pywin32/com/samples/connect.html.
Note that we cheat just a little - the Server in this demo is not created
via Normal COM - this means we can avoid registering the server.
However, the server _is_ accessed as a COM object - just the creation
is cheated on - so this is still working as a fully-fledged server.
"""

from typing import ClassVar

import pythoncom
import win32com.server.connect
import win32com.server.util

# This is the IID of the Events interface both Client and Server support.
IID_IConnectDemoEvents = pythoncom.MakeIID(
"{A4988850-49C3-11d0-AE5D-52342E000000}"
)


# The server which implements
# Create a connectable class, that has a single public method
# 'DoIt', which echos to a single sink 'DoneIt'
class ConnectableServer(win32com.server.connect.ConnectableServer):
_public_methods_: ClassVar[list[str]] = [
"DoIt",
*win32com.server.connect.ConnectableServer._public_methods_,
]
_connect_interfaces_: ClassVar = [IID_IConnectDemoEvents]

# The single public method that the client can call on us
# (ie, as a normal COM server, this exposes just this single method.
def DoIt(self, arg) -> None:
# Simply broadcast a notification.
self._BroadcastNotify(self.NotifyDoneIt, (arg,))

def NotifyDoneIt(self, interface, arg) -> None:
interface.Invoke(1000, 0, pythoncom.DISPATCH_METHOD, 1, arg)


# Here is the client side of the connection world.
# Define a COM object which implements the methods defined by the
# IConnectDemoEvents interface.
class ConnectableClient:
# This is another cheat - I _know_ the server defines the "DoneIt" event
# as DISPID==1000 - I also know from the implementation details of COM
# that the first method in _public_methods_ gets 1000.
# Normally some explicit DISPID->Method mapping is required.
_public_methods_: ClassVar = ["OnDoneIt"]

def __init__(self) -> None:
self.last_event_arg = None

# A client must implement QI, and respond to a query for the Event interface.
# In addition, it must provide a COM object (which server.util.wrap) does.
def _query_interface_(self, iid): # noqa: ANN202
import win32com.server.util

# Note that this seems like a necessary hack. I am responding to
# IID_IConnectDemoEvents but only creating an IDispatch gateway object.
if iid == IID_IConnectDemoEvents:
return win32com.server.util.wrap(self)
return None

# And here is our event method which gets called.
def OnDoneIt(self, arg) -> None:
self.last_event_arg = arg


def CheckEvent(server, client, val, verbose) -> None:
client.last_event_arg = None
server.DoIt(val)
if client.last_event_arg != val:
msg = f"Sent {val!r}, but got back {client.last_event_arg!r}"
raise RuntimeError(msg)
if verbose:
print(f"Sent and received {val!r}")


# A simple test script for all this.
# In the real world, it is likely that the code controlling the server
# will be in the same class as that getting the notifications.
def test(verbose=0) -> None:
import win32com.client.connect
import win32com.client.dynamic
import win32com.server.policy

server = win32com.client.dynamic.Dispatch(
win32com.server.util.wrap(ConnectableServer())
)
connection = win32com.client.connect.SimpleConnection()
client = ConnectableClient()
connection.Connect(server, client, IID_IConnectDemoEvents)
CheckEvent(server, client, "Hello from cx_Freeze", verbose)
CheckEvent(server, client, b"Here is a null>\x00<", verbose)
CheckEvent(server, client, "Here is a null>\x00<", verbose)
val = "test-\xe0\xf2" # 2 extended characters.
CheckEvent(server, client, val, verbose)
if verbose:
print("Everything seemed to work!")
# Aggressive memory leak checking (ie, do nothing!) :-)
# All should cleanup OK???


if __name__ == "__main__":
test(1)
1 change: 0 additions & 1 deletion tests/test_hooks_pandas.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@ def test_pandas(datafiles: Path) -> None:
assert executable.is_file()

output = run_command(datafiles, executable, timeout=10)
print(output)
lines = output.splitlines()
assert lines[0].startswith("numpy version")
assert lines[1].startswith("pandas version")
Expand Down
55 changes: 55 additions & 0 deletions tests/test_win32com.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
"""Tests for some cx_Freeze.hooks."""

from __future__ import annotations

from pathlib import Path

import pytest
from generate_samples import create_package, run_command

from cx_Freeze._compat import BUILD_EXE_DIR, EXE_SUFFIX

SAMPLES_DIR = Path(__file__).resolve().parent.parent / "samples"


pytest.importorskip("win32com", reason="Depends on extra package: pywin32")


@pytest.mark.datafiles(SAMPLES_DIR / "win32com")
def test_win32com(datafiles: Path) -> None:
"""Test that the win32com is working correctly."""
output = run_command(
datafiles, "cxfreeze --script test_win32com.py --silent"
)
executable = datafiles / BUILD_EXE_DIR / f"test_win32com{EXE_SUFFIX}"
assert executable.is_file()

output = run_command(datafiles, executable, timeout=10)
lines = output.splitlines()
assert lines[0].startswith("Sent and received 'Hello from cx_Freeze'")
assert lines[-1].startswith("Everything seemed to work!")
assert len(lines) == 5, lines


SOURCE_WIN32COM_SHELL = """
test.py
from win32com.shell import shell
import pythoncom
shortcut = pythoncom.CoCreateInstance(
shell.CLSID_ShellLink, None,
pythoncom.CLSCTX_INPROC_SERVER, shell.IID_IShellLink
)
print(shortcut)
"""


def test_win32com_shell(tmp_path: Path) -> None:
"""Test if zoneinfo hook is working correctly."""
create_package(tmp_path, SOURCE_WIN32COM_SHELL)
output = run_command(tmp_path, "cxfreeze --script test.py --silent")
executable = tmp_path / BUILD_EXE_DIR / f"test{EXE_SUFFIX}"
assert executable.is_file()
output = run_command(tmp_path, executable, timeout=10)
print(output)
lines = output.splitlines()
assert lines[0].startswith("<PyIShellLink at")

0 comments on commit 443ca14

Please sign in to comment.